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

def from_decimal(cls, dec):
        """Converts a finite Decimal instance to a rational number, exactly."""
        from decimal import Decimal
        if isinstance(dec, numbers.Integral):
            dec = Decimal(int(dec))
        elif not isinstance(dec, Decimal):
            raise TypeError(
                "%s.from_decimal() only takes Decimals, not %r (%s)" %
                (cls.__name__, dec, type(dec).__name__))
        if not dec.is_finite():
            # Catches infinities and nans.
            raise TypeError("Cannot convert %s to %s." % (dec, cls.__name__))
        sign, digits, exp = dec.as_tuple()
        digits = int(''.join(map(str, digits)))
        if sign:
            digits = -digits
        if exp >= 0:
            return cls(digits * 10 ** exp)
        else:
            return cls(digits, 10 ** -exp)

def get_forex_buy_quote(currency_code: str = 'EUR', source: str = 'FNB', order_type: str = 'buy'):
    """Get latest forex from FNB website

    """
    if source == 'FNB':
        tables = pd.read_html(
            'https://www.fnb.co.za/Controller?nav=rates.forex.list.ForexRatesList',
            index_col=1, header=0, match=currency_code)

        df = tables[0]

        types = {
            'buy': 'Bank Selling Rate',
            'sell': 'Bank Buying Rate',
        }

        exhange_rate = df.loc[currency_code, types[order_type]]

        return Decimal("%.4f" % float(exhange_rate))

def test_update_and_save_information_that_return_none(
            self, mock_get_torrent, mock_hmset,
            mock_update_and_stop_seeding_delay,
            mock_create_from_torrent
    ):
        mock_create_from_torrent.return_value = (self.file, self.file_mp4)

        mock_get_torrent.return_value = self.torrent(
            status='downloading',
            progress=Decimal('100.00'),
            ratio=Decimal('9.99'),
            rateUpload=10500,
            rateDownload=105000,
            stop=None
        )

        update_and_save_information(self.torrent_model.pk)

        self._apply_common_assertions(mock_get_torrent, mock_hmset)
        mock_get_torrent.assert_called_with(self.torrent_model.hash)
        mock_update_and_stop_seeding_delay.assert_called_with(self.torrent_model.pk)

def test_update_and_stop_seeding_that_return_apply_async(self, mock_get_torrent, mock_hset, mock_apply_async):
        mock_get_torrent.return_value = self.torrent(
            status='downloading',
            progress=Decimal('100.00'),
            ratio=Decimal('9.99'),
            rateUpload=10500,
            rateDownload=105000,
            stop=None
        )

        update_and_stop_seeding(self.torrent_model.pk)

        mock_get_torrent.assert_called_with(self.torrent_model.hash)
        mock_hset.assert_called_with(
            'torrent:{}'.format(self.torrent_model.pk), 'rate_upload', mock_get_torrent.return_value.rateUpload
        )
        mock_apply_async.assert_called_with((self.torrent_model.pk,), countdown=COUNTDOWN)

def test_update_and_stop_seeding_that_seeding_return_none(self, mock_get_torrent, mock_hset):
        mock_get_torrent.return_value = self.torrent(
            status='seeding',
            progress=Decimal('100.00'),
            ratio=Decimal('9.99'),
            rateUpload=10500,
            rateDownload=105000,
            stop=mock_stop
        )

        self.torrent_model.created = timezone.now() + timezone.timedelta(hours=-24, seconds=-1)
        self.torrent_model.save()

        self.assertIsNone(update_and_stop_seeding(self.torrent_model.pk))

        mock_get_torrent.assert_called_with(self.torrent_model.hash)
        mock_hset.assert_called_with('torrent:{}'.format(self.torrent_model.pk), 'rate_upload', 0)

def result_processor(self, dialect, coltype):
        if self.asdecimal:
            if coltype in _FLOAT_TYPES:
                return processors.to_decimal_processor_factory(
                    decimal.Decimal,
                    self._effective_decimal_return_scale)
            elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
                # pg8000 returns Decimal natively for 1700
                return None
            else:
                raise exc.InvalidRequestError(
                    "Unknown PG numeric type: %d" % coltype)
        else:
            if coltype in _FLOAT_TYPES:
                # pg8000 returns float natively for 701
                return None
            elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
                return processors.to_float
            else:
                raise exc.InvalidRequestError(
                    "Unknown PG numeric type: %d" % coltype)

def result_processor(self, dialect, coltype):
        if self.asdecimal:
            if coltype in _FLOAT_TYPES:
                return processors.to_decimal_processor_factory(
                    decimal.Decimal, self._effective_decimal_return_scale)
            elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
                # pg8000 returns Decimal natively for 1700
                return None
            else:
                raise exc.InvalidRequestError(
                    "Unknown PG numeric type: %d" % coltype)
        else:
            if coltype in _FLOAT_TYPES:
                # pg8000 returns float natively for 701
                return None
            elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
                return processors.to_float
            else:
                raise exc.InvalidRequestError(
                    "Unknown PG numeric type: %d" % coltype)

def result_processor(self, dialect, coltype):
        # XXX: does the dialect return Decimal or not???
        # if it does (in all cases), we could use a None processor as well as
        # the to_float generic processor
        if self.asdecimal:
            def process(value):
                if isinstance(value, decimal.Decimal):
                    return value
                else:
                    return decimal.Decimal(str(value))
        else:
            def process(value):
                if isinstance(value, decimal.Decimal):
                    return float(value)
                else:
                    return value
        return process

def bind_processor(self, dialect):
        super_process = super(_SybNumeric_pyodbc, self).\
            bind_processor(dialect)

        def process(value):
            if self.asdecimal and \
                    isinstance(value, decimal.Decimal):

                if value.adjusted() < -6:
                    return processors.to_float(value)

            if super_process:
                return super_process(value)
            else:
                return value
        return process

def test_enotation_decimal_large(self):
        """test exceedingly large decimals.

        """

        numbers = set([
            decimal.Decimal('4E+8'),
            decimal.Decimal("5748E+15"),
            decimal.Decimal('1.521E+15'),
            decimal.Decimal('00000000000000.1E+12'),
        ])
        self._do_test(
            Numeric(precision=25, scale=2),
            numbers,
            numbers
        )

def result_processor(self, dialect, coltype):
        if self.asdecimal:
            if dialect.supports_native_decimal:
                # we're a "numeric", DBAPI will give us Decimal directly
                return None
            else:
                util.warn('Dialect %s+%s does *not* support Decimal '
                          'objects natively, and SQLAlchemy must '
                          'convert from floating point - rounding '
                          'errors and other issues may occur. Please '
                          'consider storing Decimal numbers as strings '
                          'or integers on this platform for lossless '
                          'storage.' % (dialect.name, dialect.driver))

                # we're a "numeric", DBAPI returns floats, convert.
                return processors.to_decimal_processor_factory(
                    decimal.Decimal,
                    self.scale if self.scale is not None
                    else self._default_decimal_return_scale)
        else:
            if dialect.supports_native_decimal:
                return processors.to_float
            else:
                return None

def test_updating_cart_item(self):
        session = self.client.session

        cart = self._create_testing_cart()
        cart.session_key = session.session_key
        cart.save()

        cart_item = self._create_testing_cart_item(cart_instance=cart, product_instance=self.test_product)

        response = self.client.post(reverse('cart:update', kwargs={'pk': cart_item.pk}),
            data={'cart_item_quantity': '2'}, follow=True
        )

        messages = [msg for msg in get_messages(response.wsgi_request)]

        updated_quantity = response.context['cart'].items.first().quantity
        cart_item.quantity = updated_quantity
        cart_item.save()

        self.assertEqual(response.status_code, 200)
        self.assertEqual(cart_item.quantity, 2)
        self.assertEqual(cart_item.total_price, Decimal(cart_item.quantity * cart_item.product.price))
        self.assertEqual(messages[0].tags, 'success', 'Message type should return success type')
        self.assertEqual(messages[0].message, 'Product quantity has been updated.')

def __init__(self, value, maxterms=15, cutoff=1e-10):
        if isinstance(value, (int, float, Decimal)):
            value = Decimal(value)
            remainder = int(value)
            self.append(remainder)

            while len(self) < maxterms:
                value -= remainder
                if value > cutoff:
                    value = Decimal(1) / value
                    remainder = int(value)
                    self.append(remainder)
                else:
                    break
        elif isinstance(value, (list, tuple)):
            self.extend(value)
        else:
            raise ValueError("CFraction requires number or list")

def dsum(iterable):
    "Full precision summation using Decimal objects for intermediate values"
    # Transform (exactly) a float to m * 2 ** e where m and e are integers.
    # Convert (mant, exp) to a Decimal and add to the cumulative sum.
    # If the precision is too small for exact conversion and addition,
    # then retry with a larger precision.

    total = Decimal(0)
    for x in iterable:
        mant, exp = frexp(x)
        mant, exp = int(mant * 2.0 ** 53), exp-53
        while True:
            try:
                total += mant * Decimal(2) ** exp
                break
            except Inexact:
                getcontext().prec += 1
    return float(total)

def _o(self, obj):
        """ Translate object to json.

        Dict in python is not json, so don't be confused.
        When return object from rpc, should always use _o.
        """

        if obj == None:
            return obj
        elif isinstance(obj, Decimal):
            return str(obj)
        elif isinstance(obj, datetime):
            return obj.strftime("%Y-%m-%d %H:%M:%S")
        elif isinstance(obj, date):
            return obj.strftime("%Y-%m-%d")
        elif isinstance(obj, (list, set, tuple)):
            return self._ol(obj)
        elif isinstance(obj, dict):
            return self._od(obj)
        elif isinstance(obj, (int, str, bool, float)):
            return obj
        else:
            return self._oo(obj)

def test_normal(self):
        dataframe = pandas.DataFrame(
            [
                [0, 0.1, "a"],
                [1, 1.1, "bb"],
                [2, 2.2, "ccc"],
            ],
            columns=['id', 'value', 'name'])
        expected = TableData(
            table_name="tablename",
            header_list=['id', 'value', 'name'],
            record_list=[
                [0, Decimal('0.1'), 'a'],
                [1, Decimal('1.1'), 'bb'],
                [2, Decimal('2.2'), 'ccc'],
            ])

        assert TableData.from_dataframe(dataframe, "tablename") == expected

def default(self, o,
                dates=(datetime.datetime, datetime.date),
                times=(datetime.time,),
                textual=(decimal.Decimal, uuid.UUID, DjangoPromise),
                isinstance=isinstance,
                datetime=datetime.datetime,
                text_type=text_type):
        if isinstance(o, dates):
            if not isinstance(o, datetime):
                o = datetime(o.year, o.month, o.day, 0, 0, 0, 0)
            r = o.isoformat()
            if r.endswith("+00:00"):
                r = r[:-6] + "Z"
            return r
        elif isinstance(o, times):
            return o.isoformat()
        elif isinstance(o, textual):
            return text_type(o)
        else:
            return super(JsonEncoder, self).default(o)

def plot_training_parameters(self):
        fr = open("training_param.csv", "r")
        fr.readline()
        lines = fr.readlines()
        fr.close()
        n = 100
        nu = np.empty(n, dtype=np.float64)
        gamma = np.empty(n, dtype=np.float64)
        diff = np.empty([n, n], dtype=np.float64)
        for row in range(len(lines)):
            m = lines[row].strip().split(",")
            i = row / n
            j = row % n
            nu[i] = Decimal(m[0])
            gamma[j] = Decimal(m[1])
            diff[i][j] = Decimal(m[2])
        plt.pcolor(gamma, nu, diff, cmap="coolwarm")
        plt.title("The Difference of Guassian Classifier with Different nu, gamma")
        plt.xlabel("gamma")
        plt.ylabel("nu")
        plt.xscale("log")
        plt.yscale("log")
        plt.colorbar()
        plt.show()

def test_retrieve_existing_decimal(self):
        obj1 = DecimalSetModel.objects.create(
            values={Decimal('2.5'), Decimal('1.0'), Decimal('3.5')}
        )
        obj2 = DecimalSetModel.objects.create(
            values={Decimal('3.0'), Decimal('3.5'), Decimal('1.0')}
        )
        obj3 = DecimalSetModel.objects.create(
            values={Decimal('5.9'), Decimal('1.0'), Decimal('0.0')}
        )
        self.assertListEqual(
            list(DecimalSetModel.objects.filter(values__has=Decimal('1.0'))),
            [obj1, obj2, obj3]
        )
        self.assertListEqual(
            list(DecimalSetModel.objects.filter(values__has=Decimal('3.5'))),
            [obj1, obj2]
        )
        self.assertListEqual(
            list(DecimalSetModel.objects.filter(values__has=Decimal('9.9'))),
            []
        )

def test_escape_any(self):
        now = datetime.datetime.now()
        self.assertEqual(escape_any('foo\n\r\\bar'), r'"foo\n\r\\bar"')
        self.assertEqual(escape_any(now),
                         '"%s"^^xsd:dateTime' % now.isoformat())
        self.assertEqual(escape_any(now.date()),
                         '"%s"^^xsd:date' % now.date().isoformat())
        self.assertEqual(escape_any(now.time()),
                         '"%s"^^xsd:time' % now.time().isoformat())
        self.assertEqual(escape_any(True), 'true')
        self.assertEqual(escape_any(5), '5')
        self.assertEqual(escape_any(Decimal(5.5)), '5.5')
        self.assertEqual(escape_any(5.5), '"5.5"^^xsd:double')
        self.assertEqual(escape_any(RDFTerm("raw")), 'raw')
        self.assertEqual(escape_any(Node("subject", {})), 'subject')
        with self.assertRaises(TypeError):
            escape_any(int)

def add(self, order):
        order = {
            'id': order.get('order_id') or order['id'],
            'side': order['side'],
            'price': Decimal(order['price']),
            'size': Decimal(order.get('size') or order['remaining_size'])
        }

        if order['side'] == 'buy':
            bids = self.get_bids(order['price'])
            if bids is None:
                bids = [order]
            else:
                bids.append(order)
            self.set_bids(order['price'], bids)
        else:
            asks = self.get_asks(order['price'])
            if asks is None:
                asks = [order]
            else:
                asks.append(order)
            self.set_asks(order['price'], asks)

        if not self.whaleEnteredMarket == None and self.isWhale(order['size']):
            self.whaleEnteredMarket(order)

def match(self, order):
        size = Decimal(order['size'])
        price = Decimal(order['price'])

        if order['side'] == 'buy':
            bids = self.get_bids(price)
            if not bids:
                return
            assert bids[0]['id'] == order['maker_order_id']
            if bids[0]['size'] == size:
                self.set_bids(price, bids[1:])
            else:
                bids[0]['size'] -= size
                self.set_bids(price, bids)
        else:
            asks = self.get_asks(price)
            if not asks:
                return
            assert asks[0]['id'] == order['maker_order_id']
            if asks[0]['size'] == size:
                self.set_asks(price, asks[1:])
            else:
                asks[0]['size'] -= size
                self.set_asks(price, asks)

def _normalize_information(value, unit):
    value = decimal.Decimal(str(value))

    while value < 1:
        prev_unit = functions.previous_key(INFORMATION_UNITS, unit)
        if prev_unit is None:
            break
        value, unit = convert(value, unit, prev_unit)

    while value >= 1024:
        next_unit = functions.next_key(INFORMATION_UNITS, unit)
        if next_unit is None:
            break
        value, unit = convert(value, unit, next_unit)

    return functions.format_value(value), unit

def test_serial():
    assert s.serialize_value(None) == 'x'
    assert s.serialize_value(True) == 'true'
    assert s.serialize_value(False) == 'false'
    assert s.serialize_value(5) == 'i:5'
    assert s.serialize_value(5.0) == 'f:5.0'
    assert s.serialize_value(decimal.Decimal('5.5')) == 'n:5.5'
    assert s.serialize_value('abc') == 's:abc'
    assert s.serialize_value(b'abc') == 'b:YWJj'
    assert s.serialize_value(b'abc') == 'b:YWJj'
    assert s.serialize_value(datetime.date(2007, 12, 5)) == 'd:2007-12-05'
    assert s.serialize_value(datetime.datetime(2007, 12, 5, 12, 30, 30, tzinfo=utc)) \
        == 'dt:2007-12-05 12:30:30+00:00'
    assert s.serialize_value(datetime.time(12, 34, 56)) == 't:12:34:56'
    with raises(NotImplementedError):
        s.serialize_value(csv.reader)

def test_unserial():
    def twoway(x):
        assert s.unserialize_value(s.serialize_value(x)) == x

    twoway(None)
    twoway(True)
    twoway(False)
    twoway(5)
    twoway(5.0)
    twoway(decimal.Decimal('5.5'))
    twoway('abc')
    twoway(b'abc')
    twoway(b'abc')
    twoway(datetime.date(2007, 12, 5))
    twoway(datetime.datetime(2007, 12, 5, 12, 30, 30, tzinfo=utc))
    twoway(Z('abc'))

    with raises(ValueError):
        s.unserialize_value('zzzz:abc')

def convert_Decimal(x, encoder):
    """
    Called when an instance of U{decimal.Decimal<http://
    docs.python.org/library/decimal.html#decimal-objects>} is about to be
    encoded to an AMF stream.

    @return: If the encoder is in 'strict' mode then C{x} will be converted to
        a float. Otherwise an L{pyamf.EncodeError} with a friendly message is
        raised.
    """
    if encoder.strict is False:
        return float(x)

    raise pyamf.EncodeError(
        'Unable to encode decimal.Decimal instances as there is no way to '
        'guarantee exact conversion. Use strict=False to convert to a float.'
    )

def from_records(cls, records: List[Record]) -> 'Transaction':
        """Build a Transaction object from a list of record objects."""
        amount_item_1 = records.pop(0)
        assert isinstance(
            amount_item_1, netsgiro.records.TransactionAmountItem1)
        amount_item_2 = records.pop(0)
        assert isinstance(
            amount_item_2, netsgiro.records.TransactionAmountItem2)

        text = netsgiro.records.TransactionSpecification.to_text(records)

        return cls(
            service_code=amount_item_1.service_code,
            type=amount_item_1.transaction_type,
            number=amount_item_1.transaction_number,
            date=amount_item_1.nets_date,
            amount=Decimal(amount_item_1.amount) / 100,
            kid=amount_item_1.kid,
            reference=amount_item_2.reference,
            text=text,

            payer_name=amount_item_2.payer_name,
        )

def speedup_typing(data, delay):
    delay = decimal.Decimal(delay)
    it = iter(data["stdout"])
    try:
        e = next(it)
        while True:
            while not e[1].endswith("]# "):
                e = next(it)
            e = next(it)
            if e[1] != "#":
                continue
            while True:
                e = next(it)
                e[0] = delay
                if '\r' in e[1]:
                    break
    except StopIteration:
        pass

def get_line(self, values):
        'Return the move line for the keyword values'
        pool = Pool()
        Line = pool.get('account.move.line')
        Keyword = pool.get('account.move.template.keyword')

        line = Line()
        amount = simple_eval(decistmt(self.amount),
            functions={'Decimal': Decimal}, names=values)
        amount = self.move.company.currency.round(amount)
        if self.operation == 'debit':
            line.debit = amount
        else:
            line.credit = amount
        line.account = self.account
        if self.party:
            line.party = values.get(self.party)
        if self.description:
            line.description = self.description.format(
                **dict(Keyword.format_values(self.move, values)))
        line.tax_lines = [t.get_line(values) for t in self.taxes]

        return line

def default(self, o):
        if isinstance(o, decimal.Decimal):
            if o % 1 > 0:
                return float(o)
            else:
                return int(o)
        return super(DecimalEncoder, self).default(o)

def print_setup(task_specific_setup=None):
    """
    Print experimental setup
    :param task_specific_setup: (optional) function printing task-specific parameters
    """
    model_dict = {'lstm': 'LSTM', 'gru': 'GRU', 'skip_lstm': 'SkipLSTM', 'skip_gru': 'SkipGRU'}
    print('\n\n\tExperimental setup')
    print('\t------------------\n')
    print('\tModel: %s' % model_dict[FLAGS.model.lower()])
    print('\tNumber of layers: %d' % FLAGS.rnn_layers)
    print('\tNumber of cells: %d' % FLAGS.rnn_cells)
    print('\tBatch size: %d' % FLAGS.batch_size)
    print('\tLearning rate: %.2E' % Decimal(FLAGS.learning_rate))

    if FLAGS.grad_clip > 0:
        print('\tGradient clipping: %.1f' % FLAGS.grad_clip)
    else:
        print('\tGradient clipping: No')

    if FLAGS.model.lower().startswith('skip'):
        print('\tCost per sample: %.2E' % Decimal(FLAGS.cost_per_sample))

    if isinstance(task_specific_setup, types.FunctionType):
        print('')
        task_specific_setup()

    print('\n\n')

def convertSelectDataToCsv(result, dumpFunc=json.dumps, *args, **kargs):
    """
    Return a function that produces a generator for outputting a CSV file.

    :param result: the initial select results.
    :param dumpFunc: fallback function for dumping objects and unknown data
                     types to JSON.
    :returns: a function that outputs a generator.
    """
    class Echo(object):
        def write(self, value):
            return value

    writer = csv.writer(Echo())
    # values that are of a type in allowedTypes and not in disallowedTypes
    # should be converted by the CSV writer.  All others are converted to JSON
    # first.  The integer_types include True and False.  We may need to add
    # more type from sqlalchemy.sql.sqltypes.
    allowedTypes = six.string_types + six.integer_types + (
        float, type(None), datetime.datetime, decimal.Decimal)
    disallowedTypes = (bson.binary.Binary, )
    data = convertSelectDataToList(result)['data']

    def resultFunc():
        columns = {result['columns'][col]: col for col in result['columns']}
        columnNames = [columns[i] for i in range(len(result['fields']))]
        yield writer.writerow(csv_safe_unicode(columnNames))
        for row in data:
            row = [value if isinstance(value, allowedTypes) and
                   not isinstance(value, disallowedTypes) else
                   dumpFunc(value, check_circular=False, separators=(',', ':'),
                            sort_keys=False, default=str) for value in row]

            yield writer.writerow(csv_safe_unicode(row))

    return resultFunc

def coin_exchange(df, limit, order_type: str, bitcoin_column: str = 'volume', currency_column: str = 'value'):
    """Convert specified amount of bitcoin to currency or currency to bitcoin

    :param: order_type buy or sell
        buy exchanges currency for bitcoin
        sell exchanges bitcoins for currency

    """

    options = {
        'buy': {'from': currency_column, 'to': bitcoin_column},
        'sell': {'from': bitcoin_column, 'to': currency_column}
    }

    filtered = df.loc[df['cumulative_%s' % options[order_type]['from']] < float(limit), :]
    rows = filtered.shape[0]

    over = Decimal(df.loc[rows, 'cumulative_%s' % options[order_type]['from']]) - limit

    price = Decimal(df.loc[rows, 'price'])

    if order_type == 'buy':
        over_convert = over / price
    else:
        over_convert = over * price

    result = Decimal(df.loc[rows, 'cumulative_%s' % options[order_type]['to']]) - over_convert
    return result

def alchemyencoder(obj):
    """JSON encoder function for SQLAlchemy special classes."""
    if isinstance(obj, datetime.date):
        return obj.isoformat()
    elif isinstance(obj, decimal.Decimal):
        return float(obj)

def test_retrieve_torrent_file_that_return_has_not_finished_downloading(self):
        url = reverse('torrents:torrent-file-detail', args=[self.torrent.pk, self.file.pk])
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data.get('detail'), "The torrent hasn't finished downloading yet.")
        self.assertEqual(response.data.get('progress'), Decimal('3.93'))

def test_update_and_save_information_that_return_apply_async(self, mock_get_torrent, mock_hmset, mock_apply_async):
        mock_get_torrent.return_value = self.torrent(
            status='downloading',
            progress=Decimal('45.97'),
            ratio=Decimal('9.99'),
            rateUpload=10500,
            rateDownload=105000,
            stop=None
        )

        update_and_save_information(self.torrent_model.pk)

        self._apply_common_assertions(mock_get_torrent, mock_hmset)
        mock_get_torrent.assert_called_with(self.torrent_model.hash)
        mock_apply_async.assert_called_with((self.torrent_model.pk,), countdown=COUNTDOWN)

def save(self, **kwargs):
        self.progress = Decimal('{:.2f}'.format(self.progress))
        self.ratio = Decimal('{:.2f}'.format(self.ratio))

        return super(Torrent, self).save(**kwargs)

def test_bounty_str_no_bonus():
    r = new_report()
    b = Bounty(report=r, created_at=now(), amount=Decimal("50.00"))
    assert str(b) == "$50.00"

def test_bounty_str_with_bonus():
    r = new_report()
    b = Bounty(report=r, created_at=now(), amount=Decimal("50.00"), bonus=Decimal("5.00"))
    assert str(b) == "$50.00 + $5.00"

def test_sync_with_bounty():
    bounty = FakeBounty(amount=Decimal("50.00"))
    call_h1sync(reports=[FakeApiReport(id=1, bounties=[bounty])])
    report = Report.objects.get(id=1)
    bounties = report.bounties.all()
    assert len(bounties) == 1
    assert bounties[0].amount == Decimal("50.00")

def test_sync_with_bounty_and_bonus():
    bounty = FakeBounty(amount=Decimal("50.00"), bonus_amount=Decimal("5.00"))
    call_h1sync(reports=[FakeApiReport(id=1, bounties=[bounty])])
    assert Report.objects.get(id=1).bounties.all()[0].bonus == Decimal("5.00")

def test_sync_multiple_bounties():
    bounty1 = FakeBounty(amount=Decimal("50.00"))
    bounty2 = FakeBounty(amount=Decimal("100.00"))
    call_h1sync(reports=[FakeApiReport(id=1, bounties=[bounty1, bounty2])])
    total_bounties = sum(b.amount for b in Report.objects.get(id=1).bounties.all())
    assert total_bounties == Decimal("150.00")

def datetime2unix(d):
    try:
        if d == None:
            return None
        elif isinstance(d, datetime.datetime):
            epoch = datetime.datetime(1970, 1, 1)
        elif isinstance(d, datetime.date):
            epoch = datetime.date(1970, 1, 1)
        else:
            Log.error("Can not convert {{value}} of type {{type}}",  value= d,  type= d.__class__)

        diff = d - epoch
        return Decimal(long(diff.total_seconds() * 1000000)) / 1000000
    except Exception as e:
        Log.error("Can not convert {{value}}",  value= d, cause=e)

def get_ast_value(value):
    if isinstance(value, ast.Node):
        return value
    if isinstance(value, six.string_types):
        return ast.StringValue(value=value)
    elif isinstance(value, bool):
        return ast.BooleanValue(value=value)
    elif isinstance(value, (float, decimal.Decimal)):
        return ast.FloatValue(value=value)
    elif isinstance(value, int):
        return ast.IntValue(value=value)
    return None

def isclose(a, b, rel_tol=1e-09, abs_tol=0):
        """
        Python 2 implementation of Python 3.5 math.isclose()
        https://hg.python.org/cpython/file/v3.5.2/Modules/mathmodule.c#l1993
        """
        # sanity check on the inputs
        if rel_tol < 0 or abs_tol < 0:
            raise ValueError("tolerances must be non-negative")
        # short circuit exact equality -- needed to catch two infinities of
        # the same sign. And perhaps speeds things up a bit sometimes.
        if a == b:
            return True
        # This catches the case of two infinities of opposite sign, or
        # one infinity and one finite number. Two infinities of opposite
        # sign would otherwise have an infinite relative tolerance.
        # Two infinities of the same sign are caught by the equality check
        # above.
        if _isinf(a) or _isinf(b):
            return False
        # Cast to float to allow decimal.Decimal arguments
        if not isinstance(a, float):
            a = float(a)
        if not isinstance(b, float):
            b = float(b)
        # now do the regular computation
        # this is essentially the "weak" test from the Boost library
        diff = _fabs(b - a)
        result = ((diff <= _fabs(rel_tol * a)) or
                  (diff <= _fabs(rel_tol * b)) or
                  (diff <= abs_tol))
        return result

def round2(number, ndigits=None):
        """
        Implementation of Python 2 built-in round() function.

        Rounds a number to a given precision in decimal digits (default
        0 digits). The result is a floating point number. Values are rounded
        to the closest multiple of 10 to the power minus ndigits; if two
        multiples are equally close, rounding is done away from 0.

        ndigits may be negative.

        See Python 2 documentation:
        https://docs.python.org/2/library/functions.html?highlight=round#round
        """
        if ndigits is None:
            ndigits = 0

        if ndigits < 0:
            exponent = 10 ** (-ndigits)
            quotient, remainder = divmod(number, exponent)
            if remainder >= exponent//2 and number >= 0:
                quotient += 1
            return float(quotient * exponent)
        else:
            exponent = _decimal.Decimal('10') ** (-ndigits)

            d = _decimal.Decimal.from_float(number).quantize(
                exponent, rounding=_decimal.ROUND_HALF_UP)

            return float(d)

def test_float(self):
        self.assertEqual(json.dumps(AlternateFloat(1.0)), '1.0')
        self.assertEqual(json.dumps(AlternateFloat(-1.0)), '-1.0')
        self.assertEqual(json.loads(json.dumps({AlternateFloat(1.0): 1})), {'1.0': 1})

    # NOTE: Decimal subclasses are not supported as-is
    # def test_decimal(self):
    #     self.assertEqual(json.dumps(AlternateDecimal('1.0')), '1.0')
    #     self.assertEqual(json.dumps(AlternateDecimal('-1.0')), '-1.0')

def test_decimal_encode(self):
        for d in map(Decimal, self.NUMS):
            self.assertEqual(self.dumps(d, use_decimal=True), str(d))

def test_decimal_decode(self):
        for s in self.NUMS:
            self.assertEqual(self.loads(s, parse_float=Decimal), Decimal(s))