131 lines
3.1 KiB
Python
Executable File
131 lines
3.1 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
|
|
from os import environ
|
|
from enum import Enum
|
|
|
|
load_dotenv()
|
|
|
|
API_KEY: str | None = environ.get("API_KEY")
|
|
BASE_URL: str = "https://api.taapi.io"
|
|
|
|
|
|
class Indicator(Enum):
|
|
rsi = "rsi"
|
|
|
|
|
|
class Interval(Enum):
|
|
OneMinute = 60
|
|
FiveMinutes = 300
|
|
FifteenMinutes = 900
|
|
ThirtyMinutes = 1800
|
|
OneHour = 3600
|
|
TwoHours = 7200
|
|
FourHours = 14400
|
|
TwelveHours = 43200
|
|
OneDay = 86400
|
|
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}"
|
|
|
|
params: dict[str, str | int | bool] = {
|
|
"secret": API_KEY,
|
|
"symbol": ticker,
|
|
"interval": interval,
|
|
"type": "stocks",
|
|
"gaps": "false",
|
|
"addResultTimestamp": "true",
|
|
"backtrack": candles,
|
|
"results": "14",
|
|
}
|
|
|
|
response: requests.Response = requests.get(target_url, params=params)
|
|
|
|
return response
|
|
|
|
|
|
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 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:
|
|
date = parse_date_yyyymmdd("20250821")
|
|
indicator = Indicator.rsi
|
|
ticker = "AAPL"
|
|
response = query_indicator(ticker, indicator.value, date)
|
|
|
|
if response.status_code == 200:
|
|
data: dict[str, list[float] | list[int]] = response.json()
|
|
|
|
print(f"{indicator.value} values for {ticker} on {format_date_readable(date)}:")
|
|
|
|
for rsi_val, ts in zip(data["value"], data["timestamp"]):
|
|
dt: datetime = datetime.fromtimestamp(ts)
|
|
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|