contabilitat/commands/cash-flow-statement

#!/usr/bin/env python3
from beancount import loader
from beancount.query import query
from beancount.parser import printer
import argparse
from tabulate import tabulate
from decimal import Decimal
from beancount.core.amount import Amount, add, sub, mul
from datetime import date
from dateutil.relativedelta import relativedelta

class bcolors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKCYAN = '\033[96m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

def draw_line():
  print('─' * 30)

def get_income_val(obj, key):
  if key in obj:
    amount = obj[key].get_only_position().units
    amount = mul(amount, Decimal(-1.0))
    return amount.to_string()
  else:
    return None

def get_expense_val(obj, key):
  if key in obj:
    amount = obj[key].get_only_position().units
    return amount.to_string()
  else:
    return None

def print_expenses_table(expenses):
  table = []
  for key, expense in expenses.items():
    parts = key.split(":", 1)
    table.append([parts[1], get_expense_val(expenses, key)])
  print(tabulate(table))

def get_total_inflows(income):
  sum = 0
  for account, balance in income.items():
    sum = balance if sum == 0 else sum + balance
  if sum != 0 and sum.get_only_position() != None:
    result = sum.get_only_position().units
    return Amount(Decimal(round(result.number, 2) * Decimal(-1.0)), result.currency)
  else:
    return Amount(Decimal(0), "EUR")

def get_total_outflows(expenses, total_investments):
  sum = 0
  for account, balance in expenses.items():
    sum = balance if sum == 0 else sum + balance
  if sum != 0 and sum.get_only_position() != None:
    result = sum.get_only_position().units
    return add(Amount(Decimal(round(result.number, 2)), result.currency), total_investments)
  else:
    return total_investments

def get_total_investments(investments):
  sum = Amount(Decimal(0), "EUR")
  for inv in investments:
    sum = inv.cost_position if sum == Amount(Decimal(0), "EUR") else add(sum, inv.cost_position)
  if sum != 0 and sum != None:
    return Amount(Decimal(round(sum.number, 2)), sum.currency)
  else:
    return Amount(Decimal(0), "EUR")

def print_report(start_date, period, income, expenses, investments):
  print(f"{bcolors.BOLD}Cash Flow Statement (period={period}, start_date={start_date}){bcolors.ENDC}")
  draw_line()
  print(f"{bcolors.BOLD}Inflows{bcolors.ENDC}")
  print(f"\t{bcolors.BOLD}Income{bcolors.ENDC}")
  print(tabulate([
    ["Salari", get_income_val(income,"Income:Work:Zurich:Salari")],
    ["Tickets Restaurant", get_income_val(income,"Income:Work:Zurich:TicketsRestaurant")],
    ["Targeta Transport", get_income_val(income,"Income:Work:Zurich:TargetaTransport")],
    ["Pla pensions Empleats Zurich", get_income_val(income,"Income:Work:Zurich:DZP")],
    ["Seguro Mèdic", get_income_val(income,"Income:Work:Zurich:SeguroMedic")],
    ["Gimnàs", get_income_val(income,"Income:Work:Zurich:Gimnas")]
  ]))
  print(f"\t{bcolors.BOLD}Income from Investment{bcolors.ENDC}")
  print(tabulate([
    ["Capital Gains", get_income_val(income,"Income:Invest:R4:CapitalGains")],
    ["Dividends", get_income_val(income,"Income:Invest:R4:Dividends")],
    ["Rentabilitat Estalvis", get_income_val(income,"Income:Savings:Caixabank:RentabilitatEstalvis")]
  ]))
  print(f"\t{bcolors.BOLD}Other Inflows{bcolors.ENDC}")
  print(tabulate([
    ["Transferències", get_income_val(income,"Income:Other:Caixabank:Transferencia")],
    ["Bizum", get_income_val(income,"Income:Other:Caixabank:Bizum")],
    ["Devolucions", get_income_val(income,"Income:Other:Devolucions")]
  ]))
  print(tabulate([
    ["Total Inflows", f"{bcolors.BOLD}{get_total_inflows(income).to_string()}{bcolors.ENDC}"]
  ]))
  draw_line()
  print(f"{bcolors.BOLD}Outflows{bcolors.ENDC}")
  print_expenses_table(expenses)
  print(f"{bcolors.BOLD}Outflows from Investment{bcolors.ENDC}")
  total_investments = get_total_investments(investments)
  print(tabulate([
    ["Compra de fons i accions", total_investments.to_string()]
  ]))
  print(tabulate([
    ["Total Outflows", f"{bcolors.BOLD}{get_total_outflows(expenses, total_investments).to_string()}{bcolors.ENDC}"]
  ]))
  draw_line()
  net_cash_flow = sub(get_total_inflows(income), get_total_outflows(expenses, total_investments))
  print(tabulate([
    ["NET CASH FLOW", f"{bcolors.BOLD}{bcolors.OKGREEN if net_cash_flow.number >= 0 else bcolors.FAIL}{net_cash_flow.to_string()}{bcolors.ENDC}"]
  ]))

def get_income(entries, options, period, start_date):
  period_delta = relativedelta(months=1) if period == "monthly" else relativedelta(years=1)
  end_date = date.fromisoformat(start_date) + period_delta
  income_query = f"SELECT account, sum(position) FROM OPEN ON {start_date} CLOSE ON {end_date.isoformat()} WHERE account ~ \"Income\""
  rtypes, rrows = query.run_query(
    entries, options, income_query)
  income = {}
  for row in rrows:
    income[row.account] = row.sum_position
  return income

def get_expenses(entries, options, period, start_date):
  period_delta = relativedelta(months=1) if period == "monthly" else relativedelta(years=1)
  end_date = date.fromisoformat(start_date) + period_delta
  expenses_query = f"SELECT account, sum(position) FROM OPEN ON {start_date} CLOSE ON {end_date.isoformat()} WHERE account ~ \"Expenses\""
  rtypes, rrows = query.run_query(
    entries, options, expenses_query)
  expenses = {}
  for row in rrows:
    expenses[row.account] = row.sum_position
  return expenses

def get_investments(entries, options, period, start_date):
  period_delta = relativedelta(months=1) if period == "monthly" else relativedelta(years=1)
  end_date = date.fromisoformat(start_date) + period_delta
  expenses_query = f"SELECT account, cost(position), currency, date WHERE account ~ \"Assets:Invest:R4:\" AND NOT currency ~ '^(EUR|USD)' AND date >= {start_date} AND date < {end_date.isoformat()}"
  rtypes, rrows = query.run_query(
    entries, options, expenses_query)
  return rrows

def main():
  parser = argparse.ArgumentParser(description='Generate cash flow report')
  parser.add_argument('start_date', metavar='start_date', type=str, nargs=1,
                      help='Start date (end date will be one month after if monthly report or one year after if yearly report)')
  parser.add_argument('-p', metavar='period', type=str, choices=["monthly", "yearly"], default="monthly", required=False,
                      help='Period (monthly or yearly)')

  args = parser.parse_args()
  start_date = args.start_date[0]
  period = args.p

  filename = "ledger/main.beancount"
  entries, errors, options = loader.load_file(filename)

  if errors:
    printer.print_errors(errors)

  income = get_income(entries, options, period, start_date)
  expenses = get_expenses(entries, options, period, start_date)
  investments = get_investments(entries, options, period, start_date)
  print_report(start_date, period, income, expenses, investments)

main()