Coverage for hledger_lots/prices_yahoo.py: 30%

1import sys 

2from dataclasses import dataclass 

3from datetime import date, datetime, timedelta 

4from pathlib import Path 

5from typing import Dict, List, Tuple 

6 

7import yfinance as yf 

8from requests.exceptions import HTTPError 

9from requests_cache import CachedSession 

10 

11from .commodity_tag import CommodityDirective, CommodityTag 

12from .hl import hledger2txn 

13from .info import get_last_price 

14from .lib import get_files_comm 

15 

16@dataclass 

17class Price: 

18 name: str 

19 date: date 

20 price: float 

21 cur: str 

22 

23 

24class YahooPrices: 

25 TAG = "yahoo_ticker" 

26 

27 def __init__(self, files: Tuple[str, ...]) -> None: 

28 self.files = files 

29 self.files_comm = get_files_comm(files) 

30 

31 self.session_path = Path.home() / "yfinance.cache" 

32 self.session = CachedSession(str(self.session_path)) 

33 self.today = datetime.today() 

34 yesterday = self.today - timedelta(days=1) 

35 self.yesterday_str = yesterday.strftime("%Y-%m-%d") 

36 

37 commodity_directive = CommodityDirective(self.files) 

38 self.commodities = commodity_directive.get_commodity_tag(self.TAG) 

39 

40 def get_start_date(self, commodity: CommodityTag): 

41 txns = hledger2txn(self.files, commodity["commodity"]) 

42 qtty = sum(txn.qtty for txn in txns) 

43 if qtty == 0: 

44 print( 

45 f"; stderr: No transaction for {commodity['commodity']}. Not downloading ", 

46 file=sys.stderr, 

47 ) 

48 return 

49 

50 first_date_str = txns[0].date 

51 first_date = datetime.strptime(first_date_str, "%Y-%m-%d").date() 

52 last_market_date = get_last_price(self.files_comm, commodity["commodity"])[0] 

53 

54 if not last_market_date: 

55 last_date = first_date 

56 elif last_market_date < first_date: 

57 last_date = first_date 

58 else: 

59 last_date = last_market_date 

60 

61 start_date = last_date + timedelta(days=1) 

62 past = date.today() - start_date 

63 if past.days < 1: 

64 print(f"; stderr: No new data for {commodity}", file=sys.stderr) 

65 return 

66 

67 return start_date 

68 

69 def prices2hledger(self, prices: List[Price]): 

70 prices_list = [ 

71 f"P {price.date.strftime('%Y-%m-%d')} \"{price.name}\" {price.price} {price.cur}" 

72 for price in prices 

73 if price 

74 ] 

75 prices_str = "\n".join(prices_list) 

76 return prices_str 

77 

78 def get_prices( 

79 self, 

80 commodity: CommodityTag, 

81 start_date: str, 

82 ) -> List[Price]: 

83 ticker = yf.Ticker(commodity["value"], session=self.session) 

84 

85 try: 

86 info: Dict[str, str] = ticker.info 

87 except HTTPError: 

88 print(f"; stderr: Can't download commodity {commodity}", file=sys.stderr) 

89 return [] 

90 

91 if not start_date: 

92 return [] 

93 

94 df = ticker.history(start=start_date, end=self.yesterday_str, raise_errors=True) 

95 

96 prices = [ 

97 Price( 

98 commodity["commodity"], 

99 row[0].to_pydatetime().date(), # type:ignore 

100 row[1]["Close"], # type: ignore 

101 info["currency"], 

102 ) 

103 for row in df.iterrows() 

104 ] 

105 return prices 

106 

107 def get_commodity_prices(self, commodity: CommodityTag): 

108 start_date = self.get_start_date(commodity) 

109 if not start_date: 

110 return 

111 

112 start_date_str = start_date.strftime("%Y-%m-%d") 

113 try: 

114 prices = [ 

115 price for price in self.get_prices(commodity, start_date_str) if price 

116 ] 

117 return prices 

118 except HTTPError: 

119 print(f"; stderr: {commodity['value']} not found", file=sys.stderr) 

120 except Exception: 

121 print( 

122 f"; stderr: Nothing downloaded for {commodity['value']} between {start_date} and {self.yesterday_str}", 

123 file=sys.stderr, 

124 ) 

125 

126 def print_prices(self): 

127 if len(self.commodities) == 0: 

128 print( 

129 f"\n\n; stderr: No commodities directives with tag {self.TAG}", 

130 file=sys.stderr, 

131 ) 

132 return 

133 

134 for commodity in self.commodities: 

135 prices = self.get_commodity_prices(commodity) 

136 if prices: 

137 print("\n") 

138 hledger_prices = self.prices2hledger(prices) 

139 print(hledger_prices)