| 1 | from __future__ import annotations |
| 2 | |
| 3 | import time |
| 4 | from datetime import datetime, timedelta, timezone |
| 5 | |
| 6 | import pandas as pd |
| 7 | |
| 8 | |
| 9 | NOW = datetime.now(timezone.utc).replace(microsecond=0) |
| 10 | |
| 11 | |
| 12 | class FakeYahooTicker: |
| 13 | def __init__(self, symbol: str, scenario: str = "SUCCESS") -> None: |
| 14 | self.symbol = symbol |
| 15 | self.scenario = scenario |
| 16 | self._exchange, self._currency = identity_for(symbol) |
| 17 | annual_periods = (pd.Timestamp("2025-03-31"), pd.Timestamp("2024-03-31")) |
| 18 | quarterly_periods = (pd.Timestamp("2026-06-30"), pd.Timestamp("2026-03-31")) |
| 19 | self.income_stmt = pd.DataFrame( |
| 20 | { |
| 21 | annual_periods[0]: {"Total Revenue": 1000, "Net Income": 100, "EBITDA": 180}, |
| 22 | annual_periods[1]: {"Total Revenue": 900, "Net Income": 80, "EBITDA": 150}, |
| 23 | } |
| 24 | ) |
| 25 | self.balance_sheet = pd.DataFrame( |
| 26 | { |
| 27 | annual_periods[0]: {"Total Debt": 200, "Stockholders Equity": 800, "Total Assets": 1400}, |
| 28 | annual_periods[1]: {"Total Debt": 220, "Stockholders Equity": 720, "Total Assets": 1300}, |
| 29 | } |
| 30 | ) |
| 31 | self.cashflow = pd.DataFrame( |
| 32 | { |
| 33 | annual_periods[0]: {"Operating Cash Flow": 160, "Capital Expenditure": -40}, |
| 34 | annual_periods[1]: {"Operating Cash Flow": 130, "Capital Expenditure": -35}, |
| 35 | } |
| 36 | ) |
| 37 | self.quarterly_income_stmt = pd.DataFrame( |
| 38 | { |
| 39 | quarterly_periods[0]: {"Total Revenue": 300, "Net Income": 35, "Diluted EPS": 3.5}, |
| 40 | quarterly_periods[1]: {"Total Revenue": 275, "Net Income": 30, "Diluted EPS": 3.0}, |
| 41 | } |
| 42 | ) |
| 43 | self.quarterly_balance_sheet = pd.DataFrame( |
| 44 | { |
| 45 | quarterly_periods[0]: {"Total Debt": 190, "Stockholders Equity": 830}, |
| 46 | quarterly_periods[1]: {"Total Debt": 200, "Stockholders Equity": 800}, |
| 47 | } |
| 48 | ) |
| 49 | self.quarterly_cashflow = pd.DataFrame( |
| 50 | { |
| 51 | quarterly_periods[0]: {"Operating Cash Flow": 45}, |
| 52 | quarterly_periods[1]: {"Operating Cash Flow": 40}, |
| 53 | } |
| 54 | ) |
| 55 | |
| 56 | @property |
| 57 | def info(self): |
| 58 | if self.scenario == "TIMEOUT": |
| 59 | time.sleep(0.25) |
| 60 | if self.scenario == "UPSTREAM_FAILURE": |
| 61 | raise ConnectionError("sensitive upstream detail") |
| 62 | if self.scenario == "INVALID_INFO": |
| 63 | return ["not", "an", "object"] |
| 64 | if self.scenario == "EMPTY": |
| 65 | return {"symbol": self.symbol, "exchange": self._exchange, "currency": self._currency, "quoteType": "EQUITY"} |
| 66 | symbol = "WRONG" if self.scenario == "WRONG_SYMBOL" else self.symbol |
| 67 | exchange = "NYQ" if self.scenario == "WRONG_EXCHANGE" else self._exchange |
| 68 | currency = "USD" if self.scenario == "WRONG_CURRENCY" else self._currency |
| 69 | return { |
| 70 | "symbol": symbol, |
| 71 | "currentPrice": 250.50, |
| 72 | "regularMarketTime": int(NOW.timestamp()), |
| 73 | "currency": currency, |
| 74 | "exchange": exchange, |
| 75 | "quoteType": "EQUITY", |
| 76 | "longName": f"{self.symbol} Company", |
| 77 | "sector": "Industrials", |
| 78 | "industry": "Aerospace & Defense", |
| 79 | "trailingEps": 12.5, |
| 80 | "forwardEps": 14.0, |
| 81 | "trailingPE": 20.04, |
| 82 | "forwardPE": 17.89, |
| 83 | "priceToBook": 4.2, |
| 84 | "enterpriseToEbitda": 15.0, |
| 85 | "marketCap": float("nan") if self.scenario == "NAN_FACT" else 1000000, |
| 86 | "returnOnEquity": 0.18, |
| 87 | "revenueGrowth": 0.12, |
| 88 | "earningsGrowth": 0.15, |
| 89 | "totalDebt": 200, |
| 90 | "totalCash": 300, |
| 91 | "targetLowPrice": 220, |
| 92 | "targetMedianPrice": 275, |
| 93 | "targetMeanPrice": 280, |
| 94 | "targetHighPrice": 320, |
| 95 | "numberOfAnalystOpinions": 8, |
| 96 | "recommendationMean": 1.8, |
| 97 | "recommendationKey": "buy", |
| 98 | } |
| 99 | |
| 100 | @property |
| 101 | def news(self): |
| 102 | if self.scenario == "NO_NEWS": |
| 103 | return [] |
| 104 | return [ |
| 105 | { |
| 106 | "title": "Issuer publishes results", |
| 107 | "publisher": "Yahoo Finance", |
| 108 | "link": "https://news.example/current", |
| 109 | "providerPublishTime": int((NOW - timedelta(days=1)).timestamp()), |
| 110 | }, |
| 111 | { |
| 112 | "title": "Duplicate issuer result", |
| 113 | "publisher": "Yahoo Finance", |
| 114 | "link": "https://news.example/current", |
| 115 | "providerPublishTime": int((NOW - timedelta(days=1)).timestamp()), |
| 116 | }, |
| 117 | { |
| 118 | "title": "Old issuer result", |
| 119 | "publisher": "Yahoo Finance", |
| 120 | "link": "https://news.example/old", |
| 121 | "providerPublishTime": int((NOW - timedelta(days=31)).timestamp()), |
| 122 | }, |
| 123 | ] |
| 124 | |
| 125 | def history(self, **_kwargs): |
| 126 | if self.scenario == "UPSTREAM_FAILURE": |
| 127 | raise ConnectionError("sensitive history detail") |
| 128 | index = pd.DatetimeIndex([NOW - timedelta(days=2), NOW - timedelta(days=1), NOW]) |
| 129 | return pd.DataFrame({"Close": [248.0, float("nan"), 250.5]}, index=index) |
| 130 | |
| 131 | |
| 132 | def identity_for(symbol: str) -> tuple[str, str]: |
| 133 | if symbol.endswith(".NS"): |
| 134 | return "NSI", "INR" |
| 135 | if symbol.endswith(".AS"): |
| 136 | return "AMS", "EUR" |
| 137 | return "NMS", "USD" |
| 138 | |
| 139 | |
| 140 | def factory(scenario: str = "SUCCESS"): |
| 141 | return lambda symbol: FakeYahooTicker(symbol, scenario) |