Compute unrealized PnL day by day

unrealized.py

#!/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()