From 7909484044d0374c5980ecf33082979510577139 Mon Sep 17 00:00:00 2001 From: Giulio De Pasquale Date: Sun, 12 Oct 2025 14:14:43 +0100 Subject: [PATCH] indicators --- paperone.py | 129 ++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 115 insertions(+), 14 deletions(-) diff --git a/paperone.py b/paperone.py index 782e2ee..ce2c227 100755 --- a/paperone.py +++ b/paperone.py @@ -9,6 +9,8 @@ from typing import NoReturn, List from os import environ from enum import Enum from dataclasses import dataclass +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry load_dotenv() @@ -16,7 +18,7 @@ load_dotenv() @dataclass(frozen=True) class Indicator: endpoint: str - + params: dict[str, int] @dataclass class QueryResult: @@ -25,7 +27,28 @@ class QueryResult: class IndicatorEnum(Enum): - RSI = Indicator("rsi") + # Momentum Indicators + RSI = Indicator(endpoint="rsi", params={"period": 20}) + STOCH = Indicator(endpoint="stoch", params={"fast_k": 14, "slow_k": 3, "slow_d": 3}) + CCI = Indicator(endpoint="cci", params={"period": 20}) + + # Trend Indicators + MACD = Indicator( + endpoint="macd", + params={"fast_period": 12, "slow_period": 26, "signal_period": 9}, + ) + EMA_20 = Indicator(endpoint="ema", params={"period": 20}) + EMA_50 = Indicator(endpoint="ema", params={"period": 50}) + SMA_200 = Indicator(endpoint="sma", params={"period": 200}) + ADX = Indicator(endpoint="adx", params={"period": 14}) + + # Volatility Indicators + BBANDS = Indicator(endpoint="bbands", params={"period": 20, "stddev": 2}) + ATR = Indicator(endpoint="atr", params={"period": 14}) + + # Volume Indicators + OBV = Indicator(endpoint="obv", params={}) + VOLUME = Indicator(endpoint="volume", params={}) class Interval(Enum): @@ -45,11 +68,35 @@ 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() + self._session: requests.Session = self._create_session_with_retries() def __build_indicator_url__(self, indicator: Indicator) -> str: return f"{self._base_url}/{indicator.endpoint}" + @staticmethod + def _create_session_with_retries() -> requests.Session: + session: requests.Session = requests.Session() + + retry_strategy: Retry = Retry( + total=5, # Maximum 5 retry attempts + backoff_factor=1, # Exponential backoff: 1s, 2s, 4s, 8s, 16s + status_forcelist=[429, 500, 502, 503, 504], # Retry on these HTTP codes + allowed_methods=["GET"], # Only retry GET requests + raise_on_status=False, # Don't raise exceptions, return response + ) + + adapter: HTTPAdapter = HTTPAdapter(max_retries=retry_strategy) + + session.mount("https://", adapter) + session.mount("http://", adapter) + + return session + + def _do_get(self, url, params) -> requests.Response: + timeout = 5 + + return self._session.get(url, params=params, timeout=timeout) + def query_indicator( self, ticker: str, @@ -58,7 +105,7 @@ class TaapiClient: interval: str = "1d", results: int = 14, ) -> List[QueryResult] | None: - ret : List[QueryResult] = [] + ret: List[QueryResult] = [] backtrack_candles: int = self.__candles_to_target_date__(target_date, interval) target_url: str = self.__build_indicator_url__(indicator) @@ -73,7 +120,10 @@ class TaapiClient: "results": str(results), } - response: requests.Response = self._session.get(target_url, params=params) + if indicator.params: + params = params | indicator.params + + response = self._do_get(target_url, params) if response.status_code != 200: return None @@ -83,9 +133,43 @@ class TaapiClient: dt: datetime = datetime.fromtimestamp(ts) ret.append(QueryResult(dt, val)) - + return ret + def query_price_on_day( + self, + ticker: str, + target_date: datetime, + ) -> QueryResult | None: + backtrack_candles: int = self.__candles_to_target_date__(target_date, "1d") + target_url: str = f"{self._base_url}/price" + + params: dict[str, str | int | bool] = { + "secret": self._api_key, + "symbol": ticker, + "interval": "1d", + "type": "stocks", + "gaps": "false", + "addResultTimestamp": "true", + "backtrack": backtrack_candles, + "results": "1", + } + + response = self._do_get(target_url, params) + + if response.status_code != 200: + return None + + data = response.json() + + dt: datetime = ( + datetime.fromtimestamp(data["timestamp"][0]) + if "timestamp" in data + else target_date + ) + + return QueryResult(dt, data["value"][0]) + @staticmethod def __candles_to_target_date__( target_date: datetime, @@ -151,20 +235,37 @@ def main() -> NoReturn: exit(0) date = parse_date_yyyymmdd("20250821") - indicator: Indicator = IndicatorEnum.RSI.value - ticker = "AAPL" with TaapiClient(api_key) as client: - results = client.query_indicator(ticker, indicator, date) + # for t in ["AAPL", "NVDA", "AMD", "META", "MSFT", "GOOG"]: + for t in ["AAPL"]: + print(f"TICKER: {t}\n") + for i in IndicatorEnum: + try: + indicator_results = client.query_indicator(t, i.value, date) + except Exception as e: + # print(f"Could not retrieve data: {e}") - if not results: - print("Could not retrieve stuff") + continue - exit(1) + if not indicator_results: + # print("Could not retrieve data") - for r in [x for x in results if is_trading_day(x.datetime)]: - print(f"{format_date_readable(r.datetime)} : {r.value}") + continue + print(f"Indicator: {i}") + + trading_day_values = [ + x for x in indicator_results if is_trading_day(x.datetime) + ] + + for r in trading_day_values: + price = client.query_price_on_day(t, r.datetime) + print( + f"{format_date_readable(r.datetime)} (${price.value:.2f}) - {i.name}: {r.value:.2f}" + ) + + print("---------------") exit(0)