174 lines
4.2 KiB
Python
Executable File
174 lines
4.2 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
import requests
|
|
import math
|
|
from sys import exit
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
from typing import NoReturn, List
|
|
from os import environ
|
|
from enum import Enum
|
|
from dataclasses import dataclass
|
|
|
|
load_dotenv()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Indicator:
|
|
endpoint: str
|
|
|
|
|
|
@dataclass
|
|
class QueryResult:
|
|
datetime: datetime
|
|
value: float
|
|
|
|
|
|
class IndicatorEnum(Enum):
|
|
RSI = Indicator("rsi")
|
|
|
|
|
|
class Interval(Enum):
|
|
OneMinute = 60
|
|
FiveMinutes = 300
|
|
FifteenMinutes = 900
|
|
ThirtyMinutes = 1800
|
|
OneHour = 3600
|
|
TwoHours = 7200
|
|
FourHours = 14400
|
|
TwelveHours = 43200
|
|
OneDay = 86400
|
|
OneWeek = 604800
|
|
|
|
|
|
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()
|
|
|
|
def __build_indicator_url__(self, indicator: Indicator) -> str:
|
|
return f"{self._base_url}/{indicator.endpoint}"
|
|
|
|
def query_indicator(
|
|
self,
|
|
ticker: str,
|
|
indicator: Indicator,
|
|
target_date: datetime,
|
|
interval: str = "1d",
|
|
results: int = 14,
|
|
) -> 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,
|
|
"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)
|
|
|
|
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__(
|
|
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:
|
|
return date.weekday() not in [5, 6]
|
|
|
|
|
|
def parse_date_yyyymmdd(date_str: str) -> datetime:
|
|
return datetime.strptime(date_str, "%Y%m%d")
|
|
|
|
|
|
def format_date_readable(date: datetime) -> str:
|
|
return date.strftime("%B %d, %Y")
|
|
|
|
|
|
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 = IndicatorEnum.RSI.value
|
|
ticker = "AAPL"
|
|
|
|
with TaapiClient(api_key) as client:
|
|
results = client.query_indicator(ticker, indicator, date)
|
|
|
|
if not results:
|
|
print("Could not retrieve stuff")
|
|
|
|
exit(1)
|
|
|
|
for r in [x for x in results if is_trading_day(x.datetime)]:
|
|
print(f"{format_date_readable(r.datetime)} : {r.value}")
|
|
|
|
|
|
exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|