Coverage for hledger_lots/avg.py: 86%

1import subprocess 

2from dataclasses import dataclass 

3from datetime import date, datetime 

4from decimal import Decimal 

5from typing import List, Optional 

6 

7from hledger_lots.hl import adjust_txn 

8 

9from . import checks 

10from .lib import AdjustedTxn, CostMethodError, adjust_commodity, get_xirr 

11 

12 

13@dataclass 

14class AvgCost: 

15 date: str 

16 total_qtty: float = 0 

17 total_amount: float = 0 

18 avg_cost: float = 0 

19 

20 

21def check_sell(sell: AdjustedTxn, avg_cost: float, check: bool): 

22 if not check: 

23 return 

24 

25 decimals_price = Decimal(str(sell.price)).as_tuple().exponent 

26 decimals_avg = Decimal(str(avg_cost)).as_tuple().exponent 

27 if type(decimals_price) == int and type(decimals_avg) == int: 

28 decimals = min(abs(decimals_price), abs(decimals_avg)) 

29 else: 

30 raise ValueError("Not a decimal") 

31 

32 if abs(sell.price - avg_cost) > 10 ** (-decimals): 

33 raise CostMethodError(sell, avg_cost, sell.base_cur) 

34 pass 

35 

36 

37def get_avg_cost( 

38 txns: List[AdjustedTxn], check: bool, until: Optional[date] = None 

39) -> List[AvgCost]: 

40 if until: 

41 included_txns = [ 

42 txn 

43 for txn in txns 

44 if datetime.strptime(txn.date, "%Y-%m-%d").date() <= until 

45 ] 

46 else: 

47 included_txns = txns 

48 

49 checks.check_base_currency(included_txns) 

50 

51 total_qtty = 0 

52 total_amount = 0 

53 avg_cost = 0 

54 

55 avg_costs: List[AvgCost] = [] 

56 

57 for txn in included_txns: 

58 total_qtty += txn.qtty 

59 

60 if txn.qtty >= 0: 

61 total_amount += txn.qtty * txn.price 

62 else: 

63 check_sell(txn, avg_cost, check) 

64 total_amount += txn.qtty * avg_cost 

65 

66 avg_cost = total_amount / total_qtty if total_qtty != 0 else 0 

67 avg_costs.append(AvgCost(txn.date, total_qtty, total_amount, avg_cost)) 

68 

69 return avg_costs 

70 

71 

72def avg_sell( 

73 txns: List[AdjustedTxn], 

74 date: str, 

75 qtty: float, 

76 cur: str, 

77 cash_account: str, 

78 revenue_account: str, 

79 comm_account: str, 

80 value: float, 

81 check: bool, 

82): 

83 adj_comm = adjust_commodity(cur) 

84 checks.check_short_sell_current(txns, qtty) 

85 checks.check_base_currency(txns) 

86 checks.check_available(txns, comm_account, qtty) 

87 

88 sell_date = datetime.strptime(date, "%Y-%m-%d").date() 

89 avg_cost = get_avg_cost(txns, check) 

90 cost = avg_cost[-1].avg_cost 

91 

92 base_curr = txns[0].base_cur 

93 price = value / qtty 

94 xirr = get_xirr(price, sell_date, txns) or 0 * 100 

95 

96 txn_hl = f"""{date} Sold {cur} ; cost_method:avg_cost 

97 ; commodity:{cur}, qtty:{qtty:,.2f}, price:{price:,.2f} 

98 ; xirr:{xirr:.2f}% annual percent rate 30/360US 

99 {cash_account} {value:.2f} {base_curr} 

100 {comm_account} {qtty * -1} {adj_comm} @ {cost} {base_curr} 

101 {revenue_account}""" 

102 

103 comm = ["hledger", "-f-", "print", "--explicit"] 

104 txn_proc = subprocess.run(comm, input=txn_hl.encode(), capture_output=True) 

105 

106 txn_print: str = txn_proc.stdout.decode("utf8") 

107 return txn_print