Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
position.cost == cost):
# Check if reducing.
booking = (Booking.REDUCED
if (position.units.number * units.number) < ZERO else
Booking.AUGMENTED)
# Compute the new number ofunits.
number = position.units.number + units.number
if number == ZERO:
# If empty, delete the position.
del self[index]
position = None
else:
# Otherwise update it.
position = Position(Amount(number, units.currency), cost)
self[index] = position
break
else:
# If not found, create a new one.
if units.number == ZERO:
position = None
booking = Booking.IGNORED
else:
position = Position(units, cost)
self.append(position)
booking = Booking.CREATED
return position, booking
number_per = (weight - (cost.number_total or ZERO)) / units.number
new_cost = cost._replace(number_per=number_per)
new_pos = Position(units, new_cost)
new_posting = incomplete_posting._replace(units=new_pos.units,
cost=new_pos.cost)
else:
new_posting = None
elif missing == MissingType.COST_TOTAL:
units = incomplete_posting.units
cost = incomplete_posting.cost
assert cost.currency == weight_currency, (
"Internal error; residual currency different than missing currency.")
number_total = (weight - cost.number_per * units.number)
new_cost = cost._replace(number_total=number_total)
new_pos = Position(units, new_cost)
new_posting = incomplete_posting._replace(units=new_pos.units,
cost=new_pos.cost)
elif missing == MissingType.PRICE:
units = incomplete_posting.units
cost = incomplete_posting.cost
if cost is not None:
errors.append(InterpolationError(
incomplete_posting.meta,
"Cannot infer price for postings with units held at cost", None))
return postings, errors, True
else:
price = incomplete_posting.price
assert price.currency == weight_currency, (
"Internal error; residual currency different than missing currency.")
new_price_number = abs(weight / units.number)
if pos is not None:
# Note: In order to augment or reduce, all the fields have to match.
# Check if reducing.
booking = (Booking.REDUCED
if not same_sign(pos.units.number, units.number)
else Booking.AUGMENTED)
# Compute the new number of units.
number = pos.units.number + units.number
if number == ZERO:
# If empty, delete the position.
del self[key]
else:
# Otherwise update it.
self[key] = Position(Amount(number, units.currency), cost)
else:
# If not found, create a new one.
if units.number == ZERO:
booking = Booking.IGNORED
else:
self[key] = Position(units, cost)
booking = Booking.CREATED
return pos, booking
Args:
lot: An instance of Lot to key by.
Returns:
An pair of
found: An instance of Position, either the position that was found, or a new
Position instance that was created for this lot.
created: A boolean, true if the position had to be created.
"""
for position in self:
if position.lot == lot:
found = position
created = False
break
else:
found = Position(lot, ZERO)
self.append(found)
created = True
return found, created
def oanda_add_posting(entry, account, number, currency):
position = Position(Lot(currency, None, None), number)
posting = Posting(entry, account, position, None, None)
entry.postings.append(posting)
lot = posting.position.lot
key = Posting(None, posting.account, lot, posting.price, posting.flag)
postings_map[key] += posting.position.number
# Create a new transaction with the aggregated postings.
new_entry = Transaction(prototype_entry.fileloc,
prototype_entry.date,
prototype_entry.flag,
prototype_entry.payee,
prototype_entry.narration,
None, None, [])
for posting, number in sorted(postings_map.items()):
lot = posting.position
position = Position(lot, number)
new_entry.postings.append(
Posting(new_entry,
posting.account, position, posting.price, posting.flag))
return new_entry
# Note: we decide that it's an error to try to pad
# positions at cost; we check here that all the existing
# positions with that currency have no cost.
positions = [pos
for pos in pad_balance.get_positions()
if pos.units.currency == check_amount.currency]
for position_ in positions:
if position_.cost is not None:
pad_errors.append(
PadError(entry.meta,
("Attempt to pad an entry with cost for "
"balance: {}".format(pad_balance)),
active_pad))
# Thus our padding lot is without cost by default.
diff_position = position.Position.from_amounts(
amount.Amount(check_amount.number - balance_amount.number,
check_amount.currency))
# Synthesize a new transaction entry for the difference.
narration = ('(Padding inserted for Balance of {} for '
'difference {})').format(check_amount, diff_position)
new_entry = data.Transaction(
active_pad.meta.copy(), active_pad.date, flags.FLAG_PADDING,
None, narration, data.EMPTY_SET, data.EMPTY_SET, [])
new_entry.postings.append(
data.Posting(active_pad.account,
diff_position.units, diff_position.cost,
None, None, None))
neg_diff_position = -diff_position
new_entry.postings.append(
if pos is not None:
# Note: In order to augment or reduce, all the fields have to match.
# Check if reducing.
booking = (Booking.REDUCED
if not same_sign(pos.units.number, units.number)
else Booking.AUGMENTED)
# Compute the new number of units.
number = pos.units.number + units.number
if number == ZERO:
# If empty, delete the position.
del self[key]
else:
# Otherwise update it.
self[key] = Position(Amount(number, units.currency), cost)
else:
# If not found, create a new one.
if units.number == ZERO:
booking = Booking.IGNORED
else:
self[key] = Position(units, cost)
booking = Booking.CREATED
return pos, booking
def reduce(self, reducer, *args):
"""Reduce inventory.
Note that this returns a simple :class:`CounterInventory` with just
currencies as keys.
"""
counter = CounterInventory()
for (currency, cost), number in self.items():
pos = Position(Amount(number, currency), cost)
amount = reducer(pos, *args)
counter.add(amount.currency, amount.number)
return counter