mbleh

paperone.py

@@ -8,15 +8,18 @@ from dotenv import load_dotenv
 from typing import NoReturn
 from os import environ
 from enum import Enum
+from dataclasses import dataclass
 
 load_dotenv()
 
-API_KEY: str | None = environ.get("API_KEY")
-BASE_URL: str = "https://api.taapi.io"
+
+@dataclass(frozen=True)
+class Indicator:
+    endpoint: str
 
 
-class Indicator(Enum):
-    rsi = "rsi"
+class IndicatorEnum(Enum):
+    RSI = Indicator("rsi")
 
 
 class Interval(Enum):
@@ -32,27 +35,83 @@ class Interval(Enum):
     OneWeek = 604800
 
 
-def query_indicator(
-    ticker: str, indicator: Indicator, target_date: datetime
-) -> requests.Response:
-    interval: str = "1d"
-    candles: int = calculate_candles_to_target_date(target_date, interval)
-    target_url = f"{BASE_URL}/{indicator}"
+class TaapiClient:
+    def __init__(self, api_key: str) -> None:
+        self._api_key: str = api_key
+        self._base_url: str = "https://api.taapi.io"
+        self._session: requests.Session = requests.Session()
 
-    params: dict[str, str | int | bool] = {
-        "secret": API_KEY,
-        "symbol": ticker,
-        "interval": interval,
-        "type": "stocks",
-        "gaps": "false",
-        "addResultTimestamp": "true",
-        "backtrack": candles,
-        "results": "14",
-    }
+    def __build_indicator_url(self, indicator: Indicator) -> str:
+        return f"{self._base_url}/{indicator.endpoint}"
 
-    response: requests.Response = requests.get(target_url, params=params)
+    def query_indicator(
+        self,
+        ticker: str,
+        indicator: Indicator,
+        target_date: datetime,
+        interval: str = "1d",
+        results: int = 14,
+    ) -> requests.Response:
+        backtrack_candles: int = self._candles_to_target_date(target_date, interval)
+        target_url: str = self.__build_indicator_url(indicator)
 
-    return response
+        params: dict[str, str | int | bool] = {
+            "secret": self._api_key,
+            "symbol": ticker,
+            "interval": interval,
+            "type": "stocks",
+            "gaps": "false",
+            "addResultTimestamp": "true",
+            "backtrack": backtrack_candles,
+            "results": str(results),
+        }
+
+        response: requests.Response = self._session.get(target_url, params=params)
+
+        return response
+
+    @staticmethod
+    def _candles_to_target_date(
+        target_date: datetime,
+        interval: str = "1h",
+        current_time: datetime | None = None,
+    ) -> int:
+        if current_time is None:
+            current_time = datetime.now()
+
+        # Calculate time difference
+        time_diff: datetime = current_time - target_date
+        time_diff_seconds: float = time_diff.total_seconds()
+
+        # Parse interval to get candle duration in seconds
+        interval_map: dict[str, int] = {
+            "1m": 60,
+            "5m": 300,
+            "15m": 900,
+            "30m": 1800,
+            "1h": 3600,
+            "2h": 7200,
+            "4h": 14400,
+            "12h": 43200,
+            "1d": 86400,
+            "1w": 604800,
+        }
+
+        candle_duration_seconds: int = interval_map[interval]
+
+        # Calculate number of candles (round up)
+        num_candles: int = math.ceil(time_diff_seconds / candle_duration_seconds)
+
+        return num_candles
+
+    def close(self) -> None:
+        self._session.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.close()
 
 
 def is_trading_day(date: datetime) -> bool:
@@ -67,61 +126,36 @@ def format_date_readable(date: datetime) -> str:
     return date.strftime("%B %d, %Y")
 
 
-def calculate_candles_to_target_date(
-    target_date: datetime,
-    interval: str = "1h",
-    current_time: datetime | None = None,
-) -> int:
-    if current_time is None:
-        current_time = datetime.now()
-
-    # Calculate time difference
-    time_diff: datetime = current_time - target_date
-    time_diff_seconds: float = time_diff.total_seconds()
-
-    # Parse interval to get candle duration in seconds
-    interval_map: dict[str, int] = {
-        "1m": 60,
-        "5m": 300,
-        "15m": 900,
-        "30m": 1800,
-        "1h": 3600,
-        "2h": 7200,
-        "4h": 14400,
-        "12h": 43200,
-        "1d": 86400,
-        "1w": 604800,
-    }
-
-    candle_duration_seconds: int = interval_map[interval]
-
-    # Calculate number of candles (round up)
-    num_candles: int = math.ceil(time_diff_seconds / candle_duration_seconds)
-
-    return num_candles
-
-
 def main() -> NoReturn:
+    api_key = environ.get("API_KEY")
+
+    if not api_key:
+        print("API_KEY not set")
+
+        exit(0)
+
     date = parse_date_yyyymmdd("20250821")
-    indicator = Indicator.rsi
+    indicator: Indicator = IndicatorEnum.RSI.value
     ticker = "AAPL"
-    response = query_indicator(ticker, indicator.value, date)
 
-    if response.status_code == 200:
-        data: dict[str, list[float] | list[int]] = response.json()
+    with TaapiClient(api_key) as client:
+        response = client.query_indicator(ticker, indicator, date)
 
-        print(f"{indicator.value} values for {ticker} on {format_date_readable(date)}:")
+        if response.status_code == 200:
+            data: dict[str, list[float] | list[int]] = response.json()
 
-        for rsi_val, ts in zip(data["value"], data["timestamp"]):
-            dt: datetime = datetime.fromtimestamp(ts)
+            print(f"{indicator} values for {ticker} on {format_date_readable(date)}:")
 
-            if not is_trading_day(dt):
-                continue
+            for rsi_val, ts in zip(data["value"], data["timestamp"]):
+                dt: datetime = datetime.fromtimestamp(ts)
 
-            print(f"{dt.strftime('%Y-%m-%d %H:%M:%S')} | RSI: {rsi_val:.2f}")
-    else:
-        print(f"Error: {response.status_code}")
-        print(response.text)
+                if not is_trading_day(dt):
+                    continue
+
+                print(f"{dt.strftime('%Y-%m-%d %H:%M:%S')} | RSI: {rsi_val:.2f}")
+        else:
+            print(f"Error: {response.status_code}")
+            print(response.text)
 
     exit(0)