Coverage for hledger_lots/prompt_buy.py: 25%

1import subprocess 

2from dataclasses import dataclass 

3from typing import Optional, Tuple 

4 

5import questionary 

6 

7from . import prompt 

8from .info import LotsInfo 

9 

10 

11@dataclass 

12class BuyInfo(prompt.Tradeinfo): 

13 base_cur: str 

14 

15 

16def val_buy_qtty(answer: str): 

17 try: 

18 answer_float = float(answer) 

19 except ValueError: 

20 return "Invalid number" 

21 

22 if answer_float <= 0: 

23 return "Quantity should be positive" 

24 

25 return True 

26 

27 

28class PromptBuy(prompt.Prompt): 

29 def __init__( 

30 self, 

31 file: Tuple[str, ...], 

32 avg_cost: bool, 

33 check: bool, 

34 no_desc: Optional[str] = None, 

35 ) -> None: 

36 super().__init__(file, avg_cost, check, no_desc) 

37 all_commodities_txt = self.run_hledger_no_query_desc("commodities") 

38 self.all_commodities = [ 

39 com for com in all_commodities_txt.split("\n") if com != "" 

40 ] 

41 

42 print( 

43 """ 

44 Append a new purchase of a commodity. 

45 Use format format "y.ticker" to automatically download prices. 

46 Search on yahoo finance website the ticker for the commodity.\n""" 

47 ) 

48 print(self.initial_info) 

49 self.info = self.get_info() 

50 self.last_purchase = self.get_last_purchase(self.info) 

51 

52 def get_info(self): 

53 commodity = prompt.ask_commodities_text(self.all_commodities) 

54 info_not_found = LotsInfo( 

55 comm=commodity, 

56 cur="", 

57 qtty="0", 

58 amount="0", 

59 avg_cost="0", 

60 mkt_price=None, 

61 mkt_date=None, 

62 mkt_amount=None, 

63 mkt_profit=None, 

64 xirr=None, 

65 ) 

66 info = next( 

67 (info for info in self.infos if info["comm"] == commodity), info_not_found 

68 ) 

69 return info 

70 

71 def ask_base_cur_text(self): 

72 if self.info["cur"] == "": 

73 answer: str = prompt.custom_autocomplete( 

74 "Base Currency", self.all_commodities 

75 ).ask() 

76 else: 

77 answer = self.info["cur"] 

78 return answer 

79 

80 def ask_buy_qtty(self, info: LotsInfo): 

81 available = float(info["qtty"]) 

82 

83 answer_str: str = questionary.text( 

84 f"Quantity (available {available})", 

85 validate=val_buy_qtty, 

86 instruction="", 

87 ).ask() 

88 return answer_str 

89 

90 def ask_commodity_account(self): 

91 accts_txt = self.run_hledger("accounts") 

92 accts = [acct for acct in accts_txt.split("\n") if acct != ""] 

93 answer: str = prompt.custom_autocomplete("Commodity Account", accts).ask() 

94 return answer 

95 

96 def prompt(self): 

97 commodity = self.info["comm"] 

98 base_cur = self.ask_base_cur_text() 

99 sell_date = self.ask_date(self.last_purchase) 

100 qtty = float(self.ask_buy_qtty(self.info)) 

101 price_str = self.ask_price(self.info) 

102 

103 if price_str == "": 

104 value_str = self.ask_total(qtty, self.info) 

105 value = float(value_str) 

106 price = value / qtty 

107 else: 

108 price = float(price_str) 

109 value = qtty * price 

110 

111 cash_acct = self.ask_cash_account() 

112 

113 commodity_acct = self.ask_commodity_account() 

114 

115 result = BuyInfo( 

116 date=sell_date, 

117 quantity=qtty, 

118 commodity=commodity, 

119 base_cur=base_cur, 

120 cash_account=cash_acct, 

121 commodity_account=commodity_acct, 

122 price=price, 

123 value=value, 

124 ) 

125 return result 

126 

127 def get_hl_txn(self): 

128 buy = self.prompt() 

129 

130 txn_raw = f""" 

131{buy.date} Buy {buy.commodity} 

132 {buy.commodity_account} {buy.quantity} \"{buy.commodity}\" @ {buy.price} \"{buy.base_cur}\" 

133 {buy.cash_account} 

134""" 

135 

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

137 txn_proc = subprocess.run( 

138 comm, 

139 input=txn_raw.encode(), 

140 capture_output=True, 

141 ) 

142 if txn_proc.returncode != 0: 

143 err = txn_proc.stderr.decode("utf8") 

144 raise prompt.PromptError(err) 

145 

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

147 return txn_print