Skip to content

Instantly share code, notes, and snippets.

@wzyboy
Last active Dec 10, 2017
Embed
What would you like to do?
Compute unrealized PnL day by day
#!/usr/bin/env python
import argparse
import collections
from datetime import datetime
from beancount.loader import load_file
from beancount.ops import holdings
from beancount.core import prices
UnrealizedPnL = collections.namedtuple('UnrealizedPnL', ('date', 'symbol', 'quantity', 'book_value', 'market_value', 'pnl'))
def compute_unrealized_pnl(entries, price_map, *, symbols=None, min_date=None, max_date=None):
"""
This function computes unrealized gains and generate corresponding entries.
Args:
entries: A list of Beancount directives.
price_map: A price map returned by prices.build_price_map(entries).
symbols: A list of symbols to compute, or None, which is all symbols found in holdings.
min_date: A datetime.date object, or None, which is all dates in price db.
max_date: See min_date.
Returns:
A list of unrealized PnL entries.
"""
if not entries:
return []
# Get a list of aggregated positions
holdings_list = holdings.get_final_holdings(entries)
holdings_list = holdings.aggregate_holdings_by(
holdings_list, lambda h: (h.account, h.currency, h.cost_currency)
)
holdings_list = [h for h in holdings_list if h.currency != h.cost_currency]
if symbols is not None:
holdings_list = [h for h in holdings_list if h.currency in symbols]
# Compute unrealized PnL day by day
upnls = []
price_dates = sorted(set(pair[0] for v in price_map.values() for pair in v))
if min_date is not None:
price_dates = [d for d in price_dates if d >= min_date]
if max_date is not None:
price_dates = [d for d in price_dates if d <= max_date]
for price_date in price_dates:
for index, holding in enumerate(holdings_list):
base_quote = (holding.currency, holding.cost_currency)
_, market_price = prices.get_price(price_map, base_quote, price_date)
try:
market_value = holding.number * market_price
pnl = market_value - holding.book_value
except TypeError:
continue
upnl = UnrealizedPnL(
date=price_date, symbol=holding.currency, quantity=holding.number,
book_value=holding.book_value, market_value=market_value, pnl=pnl
)
upnls.append(upnl)
return upnls
def main():
ap = argparse.ArgumentParser()
ap.add_argument('beancount_file')
ap.add_argument('--symbols', nargs='+')
ap.add_argument('--min-date', type=lambda x: datetime.strptime(x, '%Y-%m-%d').date())
ap.add_argument('--max-date', type=lambda x: datetime.strptime(x, '%Y-%m-%d').date())
args = ap.parse_args()
entries, _, _ = load_file(args.beancount_file)
price_map = prices.build_price_map(entries)
upnls = compute_unrealized_pnl(
entries, price_map,
symbols=args.symbols, min_date=args.min_date, max_date=args.max_date
)
for upnl in upnls:
print(upnl)
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment