Python decimal 模块，localcontext() 实例源码

def represent_float_as_str(value):
    """Represent a float as a string without losing precision."""
    # In Python 2, calling str() on a float object loses precision:
    #
    # In [1]: 1.23456789012345678
    # Out[1]: 1.2345678901234567
    #
    # In [2]: 1.2345678901234567
    # Out[2]: 1.2345678901234567
    #
    # In [3]: str(1.2345678901234567)
    # Out[3]: '1.23456789012'
    #
    # The best way to ensure precision is not lost is to convert to string via Decimal:
    # https://github.com/mogui/pyorient/pull/226/files
    if not isinstance(value, float):
        raise GraphQLInvalidArgumentError(u'Attempting to represent a non-float as a float: '
                                          u'{}'.format(value))

    with decimal.localcontext() as ctx:
        ctx.prec = 20  # floats are max 80-bits wide = 20 significant digits
        return u'{:f}'.format(decimal.Decimal(value))

def __init__(self, default_position, undercut_market_by=0.01, minimum_return=1.005, market_fee=0.005):
        '''
        default_position is whether the Strategy should hold the minor currency (sell, False) or
        the major currency (buy / True) after scalping the market. For example, in a USD-CAD market
        True would mean CAD is held after scalping and False would mean USD is held after scalping.

        It is assumed that the default_position is currently being held. If not, override it by setting
        current_position after initializing the strategy.
        '''

        Strategy.__init__(self)

        self._spread_size_indicator = SpreadSize(minimum_return, market_fee)
        self.default_position = default_position
        self.current_position = default_position
        self._first_time_unprofitable = True #Prevents repeating the same message.

        # undercut_market_by is used to make the strategy's order be the next one filled on the market.
        with localcontext() as context:
            context.prec = 8
            self.undercut_market_by = Decimal(undercut_market_by)

def _load_(self, value, context):
        if isinstance(value, decimal.Decimal):
            if not self.get_options().allow_nan and not value.is_finite():
                raise ValueError()
            return value
        elif isinstance(value, text_types):
            try:
                with decimal.localcontext() as ctx:
                    ctx.traps[decimal.InvalidOperation] = 1
                    value = decimal.Decimal(value)
                    if not self.get_options().allow_nan and not value.is_finite():
                        raise ValueError()
                    return value
            except decimal.InvalidOperation:
                raise ValueError()
        elif isinstance(value, integer_types):
            return decimal.Decimal(value)
        elif isinstance(value, float):
            if not self.get_options().allow_nan:
                if math.isnan(value) or math.isinf(value):
                    raise ValueError()
            return decimal.Decimal(value)
        else:
            raise ValueError()

def float2dec(ft, decimal_digits):
    """
    Convert float (or int) to Decimal (rounding up) with the
    requested number of decimal digits.

    Arguments:
        ft (float, int): Number to convert
        decimal (int): Number of digits after decimal point

    Return:
        Decimal: Number converted to decima
    """
    with decimal.localcontext() as ctx:
        ctx.rounding = decimal.ROUND_UP
        places = decimal.Decimal(10)**(-decimal_digits)
        return decimal.Decimal.from_float(float(ft)).quantize(places)


# Sorting algos for rectangle lists

def _do_decode(self, state):
        """This is the internal function that does the JSON decoding.

        Called by the decode() method, after it has performed any Unicode decoding, etc.
        """
        buf = state.buf
        self.skipws(state)

        if buf.at_end:
            state.push_error('No value to decode')
        else:
            if state.options.decimal_context:
                dec_ctx = decimal.localcontext( state.options.decimal_context )
            else:
                dec_ctx = _dummy_context_manager

            with dec_ctx:
                state.obj = self.decodeobj(state, at_document_start=True )

            if not state.should_stop:
                # Make sure there's nothing at the end
                self.skipws(state)
                if not buf.at_end:
                    state.push_error('Unexpected text after end of JSON value')

def _format_number(cls, number):
        if isinstance(number, (Number, str)):
            try:
                value = Decimal(number)
            except InvalidOperation:
                raise ValueError('Invalid coordinate format: %r' % number)
            else:
                with localcontext() as ctx:
                    ctx.prec = 13
                    ctx.rounding = ROUND_HALF_UP
                    return value + 0
        elif isinstance(number, Decimal):
            return number
        else:
            raise TypeError('Invalid coordinates type: %r' % number)

def getPriceGross(self):
        with localcontext() as ctx:
            ctx.rounding = ROUND_HALF_EVEN

            vatPercent = Decimal(settings.INVOICE_VAT) * Decimal("0.01")
            grossPrice = self.price + (self.price * vatPercent)
            return grossPrice.quantize(Decimal("0.01"))

def getVATAmount(self):
        with localcontext() as ctx:
            ctx.rounding = ROUND_HALF_EVEN

            vatPercent = Decimal(settings.INVOICE_VAT) * Decimal("0.01")
            vatAmount = self.price * vatPercent
            return vatAmount.quantize(Decimal("0.01"))

def working(maxnum=100):
    sums = []
    for i in range(1, maxnum+1):
        rootint = int(i ** 0.5)
        if rootint ** 2 != i:
            with decimal.localcontext() as c:
                c.prec = 102
                rootstr = str(decimal.Decimal(i) ** decimal.Decimal('0.5'))
                rootstr = rootstr.replace('.', '')
                tmpsum = 0
                for j in range(100):
                    tmpsum += int(rootstr[j])
                sums.append(tmpsum)

    return sum(sums)

def set_pd_mag_from_counts(photodict,
                           c='',
                           ec='',
                           lec='',
                           uec='',
                           zp=DEFAULT_ZP,
                           sig=DEFAULT_UL_SIGMA):
    """Set photometry dictionary from a counts measurement."""
    with localcontext() as ctx:
        if lec == '' or uec == '':
            lec = ec
            uec = ec
        prec = max(
            get_sig_digits(str(c), strip_zeroes=False),
            get_sig_digits(str(lec), strip_zeroes=False),
            get_sig_digits(str(uec), strip_zeroes=False)) + 1
        ctx.prec = prec
        dlec = Decimal(str(lec))
        duec = Decimal(str(uec))
        if c != '':
            dc = Decimal(str(c))
        dzp = Decimal(str(zp))
        dsig = Decimal(str(sig))
        photodict[PHOTOMETRY.ZERO_POINT] = str(zp)
        if c == '' or float(c) < DEFAULT_UL_SIGMA * float(uec):
            photodict[PHOTOMETRY.UPPER_LIMIT] = True
            photodict[PHOTOMETRY.UPPER_LIMIT_SIGMA] = str(sig)
            photodict[PHOTOMETRY.MAGNITUDE] = str(dzp - (D25 * (dsig * duec
                                                                ).log10()))
            dnec = Decimal('10.0') ** (
                (dzp - Decimal(photodict[PHOTOMETRY.MAGNITUDE])) / D25)
            photodict[PHOTOMETRY.E_UPPER_MAGNITUDE] = str(D25 * (
                (dnec + duec).log10() - dnec.log10()))
        else:
            photodict[PHOTOMETRY.MAGNITUDE] = str(dzp - D25 * dc.log10())
            photodict[PHOTOMETRY.E_UPPER_MAGNITUDE] = str(D25 * (
                (dc + duec).log10() - dc.log10()))
            photodict[PHOTOMETRY.E_LOWER_MAGNITUDE] = str(D25 * (
                dc.log10() - (dc - dlec).log10()))

def set_pd_mag_from_flux_density(photodict,
                                 fd='',
                                 efd='',
                                 lefd='',
                                 uefd='',
                                 sig=DEFAULT_UL_SIGMA):
    """Set photometry dictionary from a flux density measurement.

    `fd` is assumed to be in microjanskys.
    """
    with localcontext() as ctx:
        if lefd == '' or uefd == '':
            lefd = efd
            uefd = efd
        prec = max(
            get_sig_digits(str(fd), strip_zeroes=False),
            get_sig_digits(str(lefd), strip_zeroes=False),
            get_sig_digits(str(uefd), strip_zeroes=False)) + 1
        ctx.prec = prec
        dlefd = Decimal(str(lefd))
        duefd = Decimal(str(uefd))
        if fd != '':
            dfd = Decimal(str(fd))
        dsig = Decimal(str(sig))
        if fd == '' or float(fd) < DEFAULT_UL_SIGMA * float(uefd):
            photodict[PHOTOMETRY.UPPER_LIMIT] = True
            photodict[PHOTOMETRY.UPPER_LIMIT_SIGMA] = str(sig)
            photodict[PHOTOMETRY.MAGNITUDE] = str(Decimal('23.9') - D25 * (
                dsig * duefd).log10())
            if fd:
                photodict[PHOTOMETRY.E_UPPER_MAGNITUDE] = str(D25 * (
                    (dfd + duefd).log10() - dfd.log10()))
        else:
            photodict[PHOTOMETRY.MAGNITUDE] = str(Decimal('23.9') - D25 *
                                                  dfd.log10())
            photodict[PHOTOMETRY.E_UPPER_MAGNITUDE] = str(D25 * (
                (dfd + duefd).log10() - dfd.log10()))
            photodict[PHOTOMETRY.E_LOWER_MAGNITUDE] = str(D25 * (
                dfd.log10() - (dfd - dlefd).log10()))

def is_profitable(self, highest_bid, lowest_ask):
        '''
        Returns: True if the spread is profitable, False if it is not.
        '''

        with localcontext() as context:
            context.prec = 8

            lowest_ask = Decimal(lowest_ask)
            highest_bid = Decimal(highest_bid)

            spread = lowest_ask - highest_bid

            return spread > self.threshold(highest_bid)

def satoshi_to_usd( satoshi ):
    '''
    Pre: satoshi is a Decimal.
    Returns: A Decimal with eight significant decimal places. The value of the satoshi in USD.
    '''
    with localcontext() as context:
        context.prec = 8
        return satoshi * current_price_of_bitcoin()

def current_price_of_bitcoin():
    '''
    Pre: coinmarketcap's api can be accessed. This will require an internet connection.
    Returns: The price of Bitcoin in USD according to coinmarketcap.com
             The variable will be a Decimal with eight decimal places.
    '''
    with localcontext() as context:
        context.prec = 8
        r = requests.get("https://api.coinmarketcap.com/v1/ticker/bitcoin/")
        return Decimal(r.json()[0]["price_usd"])

def test_decimal_extendedcontext_mismatched_infs_to_nan(self):
        # Test adding Decimal INFs with opposite sign returns NAN.
        inf = Decimal('inf')
        data = [1, 2, inf, 3, -inf, 4]
        with decimal.localcontext(decimal.ExtendedContext):
            self.assertTrue(math.isnan(statistics._sum(data)[1]))

def test_decimal_basiccontext_mismatched_infs_to_nan(self):
        # Test adding Decimal INFs with opposite sign raises InvalidOperation.
        inf = Decimal('inf')
        data = [1, 2, inf, 3, -inf, 4]
        with decimal.localcontext(decimal.BasicContext):
            self.assertRaises(decimal.InvalidOperation, statistics._sum, data)

def to_wei(number, unit):
    """
    Takes a number of a unit and converts it to wei.
    """
    if unit.lower() not in units:
        raise ValueError(
            "Unknown unit.  Must be one of {0}".format('/'.join(units.keys()))
        )

    if is_integer(number) or is_string(number):
        d_number = decimal.Decimal(value=number)
    elif isinstance(number, float):
        d_number = decimal.Decimal(value=str(number))
    elif isinstance(number, decimal.Decimal):
        d_number = number
    else:
        raise TypeError("Unsupported type.  Must be one of integer, float, or string")

    s_number = str(number)

    if d_number == 0:
        return 0

    unit_value = units[unit.lower()]

    if d_number < 1 and '.' in s_number:
        with localcontext() as ctx:
            multiplier = len(s_number) - s_number.index('.') - 1
            ctx.prec = multiplier
            d_number = decimal.Decimal(value=number, context=ctx) * 10**multiplier
        unit_value /= 10**multiplier

    result_value = d_number * unit_value

    if result_value < MIN_WEI or result_value > MAX_WEI:
        raise ValueError("Resulting wei value must be between 1 and 2**256 - 1")

    return int(result_value)

def to_decimal(self):
        """Returns an instance of :class:`decimal.Decimal` for this
        :class:`Decimal128`.
        """
        high = self.__high
        low = self.__low
        sign = 1 if (high & _SIGN) else 0

        if (high & _SNAN) == _SNAN:
            return decimal.Decimal((sign, (), 'N'))
        elif (high & _NAN) == _NAN:
            return decimal.Decimal((sign, (), 'n'))
        elif (high & _INF) == _INF:
            return decimal.Decimal((sign, (0,), 'F'))

        if (high & _EXPONENT_MASK) == _EXPONENT_MASK:
            exponent = ((high & 0x1fffe00000000000) >> 47) - _EXPONENT_BIAS
            return decimal.Decimal((sign, (0,), exponent))
        else:
            exponent = ((high & 0x7fff800000000000) >> 49) - _EXPONENT_BIAS

        arr = bytearray(15)
        mask = 0x00000000000000ff
        for i in range(14, 6, -1):
            arr[i] = (low & mask) >> ((14 - i) << 3)
            mask = mask << 8

        mask = 0x00000000000000ff
        for i in range(6, 0, -1):
            arr[i] = (high & mask) >> ((6 - i) << 3)
            mask = mask << 8

        mask = 0x0001000000000000
        arr[0] = (high & mask) >> 48

        # Have to convert bytearray to bytes for python 2.6.
        digits = [int(digit) for digit in str(_from_bytes(bytes(arr), 'big'))]

        with decimal.localcontext(_DEC128_CTX) as ctx:
            return ctx.create_decimal((sign, digits, exponent))

def truncate(number, places):
    if not isinstance(places, int):
        raise ValueError("Decimal places must be an integer.")
    if places < 1:
        raise ValueError("Decimal places must be at least 1.")
    # If you want to truncate to 0 decimal places, just do int(number).

    with localcontext() as context:
        context.rounding = ROUND_DOWN
        exponent = Decimal(str(math.pow(10, - places)))
        return Decimal(str(number)).quantize(exponent)

def test_decimal_mismatched_infs_to_nan(self):
        # Test adding Decimal INFs with opposite sign returns NAN.
        inf = Decimal('inf')
        data = [1, 2, inf, 3, -inf, 4]
        with decimal.localcontext(decimal.ExtendedContext):
            self.assertTrue(math.isnan(statistics._sum(data)))

def test_decimal_mismatched_infs_to_nan(self):
        # Test adding Decimal INFs with opposite sign raises InvalidOperation.
        inf = Decimal('inf')
        data = [1, 2, inf, 3, -inf, 4]
        with decimal.localcontext(decimal.BasicContext):
            self.assertRaises(decimal.InvalidOperation, statistics._sum, data)

def process_order_book(self, highest_bid, lowest_ask):
        '''
        Pre: Traders keep track of their balance / assets and do not attempt a trade when they do not
             have the balance to buy with or the assets to sell.
        '''

        # Financial calculations need accurate decimals
        with localcontext() as context:
            context.prec = 8

            # The spread has to be calculated using the values that the strategy will try to use, not what
            # is already being used.
            highest_bid = Decimal(highest_bid)+self.undercut_market_by
            lowest_ask = Decimal(lowest_ask)-self.undercut_market_by

            if self._spread_size_indicator.is_profitable(highest_bid, lowest_ask):

                # Changing current_position causes the trader to vacillate between buying and selling.
                # Traders will not enter a position when the trader does not have the balance/assets
                # so this causes the Trader to wait until their position is exited before they enter the
                # market again - with the opposite position.

                if self.current_position:
                    self.notify_observers(False, lowest_ask)
                    self.current_position = False
                else:
                    self.notify_observers(True, highest_bid)
                    self.current_position = True
                self._first_time_unprofitable = True

            else:

                # Not profitable = hold default position.
                if self.default_position == DefaultPosition.HOLD:
                    if self._first_time_unprofitable:
                        print("Spread is not profitable. Holding.")
                    self.notify_observers(None, -1) # market_value is irrelevant so it will be set to -1
                else:
                    if self.default_position == DefaultPosition.BUY:
                        if self._first_time_unprofitable:
                            print("Spread is not profitable. Holding major currency.")
                        self.notify_observers(None, lowest_ask)
                    elif self.default_position == DefaultPosition.SELL:
                        if self._first_time_unprofitable:
                            print("Spread is not profitable. Holding minor currency.")
                        self.notify_observers(None, highest_bid)
                    self.current_position = self.default_position
                self._first_time_unprofitable = False

def __init__(self, options, percentage_to_trade=1, start_by_buying=True, starting_amount=100):
        '''
        Pre: options is an instance of QuadrigaOptions and must have pair and ticker set.
        Post: self.is_test = True until authenticate() is called
        '''

        # Trader is in test mode by default.
        # minimum_trade is the minimum amount of assets that can be sold on a trade.
        Trader.__init__(self, True, options.minimum_trade, options.ticker)

        # Will be set to a number (order ID) when an order is placed.
        self._waiting_for_order_to_fill = None

        # In test mode: Is used to prevent the same transaction from being counted twice.
        self._last_simulation_transaction_check = 0

        # In test mode: tracks how much the trader's order has been filled.
        self._expecting_simulation_balance = 0
        self._expecting_simulation_assets = 0
        self._filled_simulation_balance = 0
        self._filled_simulation_assets = 0

        # Used when aborting to determine if any positions need to be closed.
        self._active_buy_order = False
        self._active_sell_order = False

        self.major_currency = options.major_currency
        self.minor_currency = options.minor_currency
        self.percentage_to_trade = Decimal(percentage_to_trade)
        self.amount_precision = options.amount_precision
        self.price_precision = options.price_precision
        self.start_by_buying = start_by_buying

        with localcontext() as context:
            context.prec = 8

            if self.start_by_buying:
                self.balance = Decimal(starting_amount)
                self.assets = Decimal(0)
            else:
                self.balance = Decimal(0)
                self.assets = Decimal(starting_amount)
            self.post_fee = Decimal(1) - Decimal(options.fee)

def __init__(self, trading_pair, percentage_to_trade=1, start_by_buying=True, starting_amount=1):
        '''
        Post: self.is_test = True until authenticate() is called
        '''

        # Split up the trading pair for currency specific options and for console output.
        self.market_ticker = trading_pair
        split_pair = trading_pair.split("_")
        self.major_currency = split_pair[0]
        self.minor_currency = split_pair[1]

        # Trader is in test mode by default.
        # minimum_trade is the minimum amount of assets that can be sold on a trade.
        Trader.__init__(self, True, minimum_trade_for[self.minor_currency])

        # Will be set to a number (order ID) when an order is placed.
        self._waiting_for_order_to_fill = None

        # In test mode: Is used to prevent the same transaction from being counted twice.
        self._last_simulation_transaction_check = 0

        # In test mode: tracks how much the trader's order has been filled.
        self._expecting_simulation_balance = 0
        self._expecting_simulation_assets = 0
        self._filled_simulation_balance = 0
        self._filled_simulation_assets = 0

        # Needs to be None or the program will crash in simulation mode when it
        # checks if the bot was shut off manually (a live-only feature)
        self.emergency_shutdown_id = None

        # Used when aborting to determine if any positions need to be closed.
        self._active_buy_order = False
        self._active_sell_order = False

        self.percentage_to_trade = Decimal(percentage_to_trade)
        self.amount_precision = 8
        self.price_precision = 8
        self.start_by_buying = start_by_buying

        with localcontext() as context:
            context.prec = 8

            if self.start_by_buying:
                self.balance = Decimal(starting_amount)
                self.assets = Decimal(0)
            else:
                self.balance = Decimal(0)
                self.assets = Decimal(starting_amount)
            self.post_fee = Decimal(1) - cryptopia_fee

def _decimal_to_128(value):
    """Converts a decimal.Decimal to BID (high bits, low bits).

    :Parameters:
      - `value`: An instance of decimal.Decimal
    """
    with decimal.localcontext(_DEC128_CTX) as ctx:
        value = ctx.create_decimal(value)

    if value.is_infinite():
        return _NINF if value.is_signed() else _PINF

    sign, digits, exponent = value.as_tuple()

    if value.is_nan():
        if digits:
            raise ValueError("NaN with debug payload is not supported")
        if value.is_snan():
            return _NSNAN if value.is_signed() else _PSNAN
        return _NNAN if value.is_signed() else _PNAN

    significand = int("".join([str(digit) for digit in digits]))
    bit_length = _bit_length(significand)

    high = 0
    low = 0
    for i in range(min(64, bit_length)):
        if significand & (1 << i):
            low |= 1 << i

    for i in range(64, bit_length):
        if significand & (1 << i):
            high |= 1 << (i - 64)

    biased_exponent = exponent + _EXPONENT_BIAS

    if high >> 49 == 1:
        high = high & 0x7fffffffffff
        high |= _EXPONENT_MASK
        high |= (biased_exponent & 0x3fff) << 47
    else:
        high |= biased_exponent << 49

    if sign:
        high |= _SIGN

    return high, low

def binomial_factorial(self):
        r"""
        Implementation of the binomial coefficient computation. Not meant for actual computation
        as the other methods available are more efficient.

        Parameters
        ----------
        n : int
            Number of possibilities
        k : int
            number of unordered outcomes

        Returns
        -------
        float
            The binomial coefficient

        Notes
        -----
        The binomial coefficient equation (in compact form) is defined as:

        .. math::

            \binom{n}{k} = \frac{n!}{k!(n-k)!} \qquad 0 \leq k \leq n

        References
        ----------
        Binomial coefficient. (2017, April 17). In Wikipedia, The Free Encyclopedia.
            From https://en.wikipedia.org/w/index.php?title=Binomial_coefficient&oldid=775905810

        Press, W., Teukolsky, S., Vetterling, W., & Flannery, B. (2007). Numerical recipes (3rd ed.).
            Cambridge: Cambridge University Press.

        Weisstein, Eric W. "Binomial Coefficient." From MathWorld--A Wolfram Web Resource.
            http://mathworld.wolfram.com/BinomialCoefficient.html

        """
        nk = np.minimum(self.n, self.n - self.k)

        if nk >= 100:
            with localcontext() as ctx:
                ctx.prec = 50
                bico = Decimal(factorial(self.n)) / (Decimal(factorial(self.k)) * Decimal(factorial(nk)))
        else:
            bico = float(factorial(self.n)) / float(factorial(self.k) * factorial(nk))

        return bico

def ramanujan(n, prec=100):
    r"""
    Approximates the factorial :math:`n!` given an integer :math:`n` using Ramanujan's formula.
    Ramanujan's formula is just as or more accurate than several other factorial approximation
    formulas.

    Parameters
    ----------
    n
        Integer to approximate factorial
    prec
        Defines level of precision for factorials over 100. Default 100. Optional

    Returns
    -------
    int or Decimal
        Factorial of :math:`n` as approximated by Ramanujan's formula.

    Notes
    -----
    Ramanujan's formula is another factorial approximation method known for its accuracy
    in comparison to other factorial approximation approaches including Stirling's and
    Gosper's approximations. Ramanujan's formula is defined as:

    .. math::

        n! \approx \sqrt{\pi} \left(\frac{n}{e}\right)^n \sqrt[6]{8n^3 + 4n^2 + n + \frac{1}{30}}

    Examples
    --------
    >>> ramanujan(10)
    3628800.3116126074
    >>> ramanujan(5)
    120.00014706585664

    References
    ----------
    Mortici, Cristinel. On Gosper's Formula for the Gamma Function. Valahia University of Targoviste,
        Department of Mathematics. Retrieved from http://files.ele-math.com/articles/jmi-05-53.pdf

    """
    if n != np.floor(n):
        n = np.floor(n)

    if n >= 100:
        with localcontext() as ctx:
            ctx.prec = prec
            f = Decimal(
                np.sqrt(np.pi) * n ** n * np.exp(-n) * (8. * n ** 3. + 4. * n ** 2. + n + 1. / 30.) ** (1. / 6.))
    else:
        f = np.sqrt(np.pi) * n ** n * np.exp(-n) * (8. * n ** 3. + 4. * n ** 2. + n + (1. / 30.)) ** (1. / 6.)

    return f