diff --git a/paperone.py b/paperone.py index e0a743c..782e2ee 100755 --- a/paperone.py +++ b/paperone.py @@ -5,7 +5,7 @@ import math from sys import exit from datetime import datetime from dotenv import load_dotenv -from typing import NoReturn +from typing import NoReturn, List from os import environ from enum import Enum from dataclasses import dataclass @@ -18,6 +18,12 @@ class Indicator: endpoint: str +@dataclass +class QueryResult: + datetime: datetime + value: float + + class IndicatorEnum(Enum): RSI = Indicator("rsi") @@ -41,7 +47,7 @@ class TaapiClient: self._base_url: str = "https://api.taapi.io" self._session: requests.Session = requests.Session() - def __build_indicator_url(self, indicator: Indicator) -> str: + def __build_indicator_url__(self, indicator: Indicator) -> str: return f"{self._base_url}/{indicator.endpoint}" def query_indicator( @@ -51,9 +57,10 @@ class TaapiClient: 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) + ) -> List[QueryResult] | None: + ret : List[QueryResult] = [] + backtrack_candles: int = self.__candles_to_target_date__(target_date, interval) + target_url: str = self.__build_indicator_url__(indicator) params: dict[str, str | int | bool] = { "secret": self._api_key, @@ -68,10 +75,19 @@ class TaapiClient: response: requests.Response = self._session.get(target_url, params=params) - return response + if response.status_code != 200: + return None + + data: dict[str, list[float] | list[int]] = response.json() + for val, ts in zip(data["value"], data["timestamp"]): + dt: datetime = datetime.fromtimestamp(ts) + + ret.append(QueryResult(dt, val)) + + return ret @staticmethod - def _candles_to_target_date( + def __candles_to_target_date__( target_date: datetime, interval: str = "1h", current_time: datetime | None = None, @@ -139,23 +155,16 @@ def main() -> NoReturn: ticker = "AAPL" with TaapiClient(api_key) as client: - response = client.query_indicator(ticker, indicator, date) + results = client.query_indicator(ticker, indicator, date) - if response.status_code == 200: - data: dict[str, list[float] | list[int]] = response.json() + if not results: + print("Could not retrieve stuff") - print(f"{indicator} values for {ticker} on {format_date_readable(date)}:") + exit(1) - for rsi_val, ts in zip(data["value"], data["timestamp"]): - dt: datetime = datetime.fromtimestamp(ts) + for r in [x for x in results if is_trading_day(x.datetime)]: + print(f"{format_date_readable(r.datetime)} : {r.value}") - 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)