aboutsummaryrefslogtreecommitdiffstats
path: root/beancount_extras_kris7t/plugins/selective_implicit_prices.py
diff options
context:
space:
mode:
Diffstat (limited to 'beancount_extras_kris7t/plugins/selective_implicit_prices.py')
-rw-r--r--beancount_extras_kris7t/plugins/selective_implicit_prices.py157
1 files changed, 157 insertions, 0 deletions
diff --git a/beancount_extras_kris7t/plugins/selective_implicit_prices.py b/beancount_extras_kris7t/plugins/selective_implicit_prices.py
new file mode 100644
index 0000000..07dc893
--- /dev/null
+++ b/beancount_extras_kris7t/plugins/selective_implicit_prices.py
@@ -0,0 +1,157 @@
1"""This plugin synthesizes Price directives for all Postings with a price or
2directive or if it is an augmenting posting, has a cost directive.
3
4Price directives will be synthesized only for commodities with the
5implicit-prices: TRUE metadata set.
6"""
7__copyright__ = "Copyright (C) 2015-2017 Martin Blais, " + \
8 "2020 Kristóf Marussy <kristof@marussy.com>"
9__license__ = "GNU GPLv2"
10
11import collections
12from typing import List, Tuple, Set
13
14from beancount.core.data import Commodity, Entries, Transaction
15from beancount.core import data
16from beancount.core import amount
17from beancount.core import inventory
18from beancount.core.position import Cost
19
20__plugins__ = ('add_implicit_prices',)
21
22
23ImplicitPriceError = collections.namedtuple('ImplicitPriceError', 'source message entry')
24
25
26METADATA_FIELD = "__implicit_prices__"
27IMPLICIT_PRICES_META = "implicit-prices"
28
29
30def fetch_commodities(entries: Entries) -> Tuple[Set[str], List[ImplicitPriceError]]:
31 commodities: Set[str] = set()
32 errors: List[ImplicitPriceError] = []
33 for entry in entries:
34 if isinstance(entry, Commodity):
35 implicit_prices = entry.meta.get(IMPLICIT_PRICES_META, False)
36 if not isinstance(implicit_prices, bool):
37 errors.append(ImplicitPriceError(
38 entry.meta,
39 f'{IMPLICIT_PRICES_META} must be Boolean, got {implicit_prices} instead',
40 entry))
41 if implicit_prices:
42 commodities.add(entry.currency)
43 return commodities, errors
44
45
46def add_implicit_prices(entries: Entries,
47 unused_options_map) -> Tuple[Entries, List[ImplicitPriceError]]:
48 """Insert implicitly defined prices from Transactions.
49
50 Explicit price entries are simply maintained in the output list. Prices from
51 postings with costs or with prices from Transaction entries are synthesized
52 as new Price entries in the list of entries output.
53
54 Args:
55 entries: A list of directives. We're interested only in the Transaction instances.
56 unused_options_map: A parser options dict.
57 Returns:
58 A list of entries, possibly with more Price entries than before, and a
59 list of errors.
60 """
61 new_entries: Entries = []
62 errors: List[ImplicitPriceError] = []
63
64 commodities, fetch_errors = fetch_commodities(entries)
65 errors.extend(fetch_errors)
66
67 # A dict of (date, currency, cost-currency) to price entry.
68 new_price_entry_map = {}
69
70 balances = collections.defaultdict(inventory.Inventory)
71 for entry in entries:
72 # Always replicate the existing entries.
73 new_entries.append(entry)
74
75 if isinstance(entry, Transaction):
76 # Inspect all the postings in the transaction.
77 for posting in entry.postings:
78 units = posting.units
79 if units.currency not in commodities:
80 continue
81
82 cost = posting.cost
83
84 # Check if the position is matching against an existing
85 # position.
86 _, booking = balances[posting.account].add_position(posting)
87
88 # Add prices when they're explicitly specified on a posting. An
89 # explicitly specified price may occur in a conversion, e.g.
90 # Assets:Account 100 USD @ 1.10 CAD
91 # or, if a cost is also specified, as the current price of the
92 # underlying instrument, e.g.
93 # Assets:Account 100 HOOL {564.20} @ {581.97} USD
94 if posting.price is not None:
95 meta = data.new_metadata(entry.meta["filename"], entry.meta["lineno"])
96 meta[METADATA_FIELD] = "from_price"
97 price_entry = data.Price(meta, entry.date,
98 units.currency,
99 posting.price)
100
101 # Add costs, when we're not matching against an existing
102 # position. This happens when we're just specifying the cost,
103 # e.g.
104 # Assets:Account 100 HOOL {564.20}
105 elif (cost is not None and
106 booking != inventory.MatchResult.REDUCED):
107 # TODO(blais): What happens here if the account has no
108 # booking strategy? Do we end up inserting a price for the
109 # reducing leg? Check.
110 meta = data.new_metadata(entry.meta["filename"], entry.meta["lineno"])
111 meta[METADATA_FIELD] = "from_cost"
112 if isinstance(cost, Cost):
113 price_entry = data.Price(meta, entry.date,
114 units.currency,
115 amount.Amount(cost.number, cost.currency))
116 else:
117 errors.append(
118 ImplicitPriceError(
119 entry.meta,
120 f'Expected {entry} to have a Cost, got {cost} instead',
121 entry))
122 price_entry = None
123 else:
124 price_entry = None
125
126 if price_entry is not None:
127 key = (price_entry.date,
128 price_entry.currency,
129 price_entry.amount.number, # Ideally should be removed.
130 price_entry.amount.currency)
131 try:
132 new_price_entry_map[key]
133
134 # Do not fail for now. We still have many valid use
135 # cases of duplicate prices on the same date, for
136 # example, stock splits, or trades on two dates with
137 # two separate reported prices. We need to figure out a
138 # more elegant solution for this in the long term.
139 # Keeping both for now. We should ideally not use the
140 # number in the de-dup key above.
141 #
142 # dup_entry = new_price_entry_map[key]
143 # if price_entry.amount.number == dup_entry.amount.number:
144 # # Skip duplicates.
145 # continue
146 # else:
147 # errors.append(
148 # ImplicitPriceError(
149 # entry.meta,
150 # "Duplicate prices for {} on {}".format(entry,
151 # dup_entry),
152 # entry))
153 except KeyError:
154 new_price_entry_map[key] = price_entry
155 new_entries.append(price_entry)
156
157 return new_entries, errors