#!/usr/bin/env python3 import os import json import time from datetime import datetime, timedelta, timezone from pathlib import Path from typing import List, Dict, Any import requests POLYGON_BASE_URL = "https://api.polygon.io/v2/aggs/ticker" SYMBOLS = ["SPY", "QQQ"] TIMEFRAMES = { "daily": {"multiplier": 1, "timespan": "day"}, "h1": {"multiplier": 1, "timespan": "hour"}, "m15": {"multiplier": 15, "timespan": "minute"}, "m5": {"multiplier": 5, "timespan": "minute"}, "m1": {"multiplier": 1, "timespan": "minute"}, } # 1-Minuten-Daten werden in 120‑Tage‑Blöcken geladen CHUNK_DAYS_1M = 120 ROOT_DIR = Path(os.path.expanduser("~/kleiner")) DATA_DIR = ROOT_DIR / "data" / "market" def load_api_key() -> str: """Lädt den Polygon‑API‑Key aus der Umgebung.""" env_path = ROOT_DIR / ".env" if env_path.exists(): try: from dotenv import load_dotenv # pylint: disable=import-error load_dotenv(env_path) except ImportError: pass # Wenn python-dotenv nicht installiert ist, wird API-Key aus Umgebung gelesen api_key = os.getenv("POLYGON_API_KEY") if not api_key: raise RuntimeError( "POLYGON_API_KEY ist nicht gesetzt. Bitte trage ihn in .env ein." ) return api_key def iso_utc_from_ms(ms: int) -> str: dt = datetime.fromtimestamp(ms / 1000, tz=timezone.utc) return dt.replace(microsecond=0).isoformat().replace("+00:00", "Z") def fetch_polygon_range( symbol: str, multiplier: int, timespan: str, from_date: str, to_date: str, api_key: str, retries: int = 3, pause: int = 5, ) -> List[Dict[str, Any]]: """ Holt eine Range aggregierter Bars von Polygon. from_date und to_date im Format YYYY-MM-DD. """ url = f"{POLYGON_BASE_URL}/{symbol}/range/{multiplier}/{timespan}/{from_date}/{to_date}" params = { "adjusted": "true", "sort": "asc", "limit": 50000, "apiKey": api_key, } for attempt in range(1, retries + 1): try: resp = requests.get(url, params=params, timeout=30) if resp.status_code == 429: time.sleep(pause * attempt) continue if resp.status_code >= 500: time.sleep(pause * attempt) continue resp.raise_for_status() data = resp.json() results = data.get("results") or [] bars: List[Dict[str, Any]] = [] for r in results: if "t" not in r: continue bars.append( { "t": iso_utc_from_ms(r["t"]), "o": r.get("o"), "h": r.get("h"), "l": r.get("l"), "c": r.get("c"), "v": r.get("v"), } ) return bars except Exception: if attempt < retries: time.sleep(pause * attempt) else: raise return [] def chunked_fetch_1m( symbol: str, from_date: datetime, to_date: datetime, api_key: str, ) -> List[Dict[str, Any]]: """ Holt 1-Minuten-Daten in 120‑Tage‑Blöcken. """ all_bars: List[Dict[str, Any]] = [] current_start = from_date while current_start <= to_date: current_end = min(current_start + timedelta(days=CHUNK_DAYS_1M - 1), to_date) bars = fetch_polygon_range( symbol=symbol, multiplier=1, timespan="minute", from_date=current_start.date().isoformat(), to_date=current_end.date().isoformat(), api_key=api_key, ) all_bars.extend(bars) current_start = current_end + timedelta(days=1) time.sleep(1) # Schonung der Rate Limits # Deduplizieren nach Zeitstempel dedup: Dict[str, Dict[str, Any]] = {} for b in all_bars: dedup[b["t"]] = b merged = [dedup[t] for t in sorted(dedup.keys())] return merged def fetch_full_history_for_symbol(symbol: str, days_back: int = 730) -> Dict[str, Any]: """ Holt volle Historie (z. B. 2 Jahre) für alle Timeframes eines Symbols. """ api_key = load_api_key() end_date = datetime.now(timezone.utc).date() start_date = end_date - timedelta(days=days_back) result: Dict[str, Any] = { "symbol": symbol, "timezone": "UTC", "last_update": datetime.now(timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z"), "daily": [], "h1": [], "m15": [], "m5": [], "m1": [], } for key in ["daily", "h1", "m15", "m5"]: cfg = TIMEFRAMES[key] bars = fetch_polygon_range( symbol=symbol, multiplier=cfg["multiplier"], timespan=cfg["timespan"], from_date=start_date.isoformat(), to_date=end_date.isoformat(), api_key=api_key, ) result[key] = bars time.sleep(1) result["m1"] = chunked_fetch_1m( symbol, datetime.combine(start_date, datetime.min.time(), tzinfo=timezone.utc), datetime.combine(end_date, datetime.min.time(), tzinfo=timezone.utc), api_key, ) return result def save_symbol_history(symbol: str, data: Dict[str, Any]) -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) path = DATA_DIR / f"{symbol.lower()}.json" with path.open("w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def load_symbol_history(symbol: str) -> Dict[str, Any]: path = DATA_DIR / f"{symbol.lower()}.json" if not path.exists(): raise FileNotFoundError(f"Keine History-Datei für {symbol} unter {path}") with path.open("r", encoding="utf-8") as f: return json.load(f) def merge_bars(existing: List[Dict[str, Any]], new: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Merged bestehende + neue Bars, deduziert nach 't', sortiert aufsteigend. """ merged: Dict[str, Dict[str, Any]] = {b["t"]: b for b in existing} for b in new: merged[b["t"]] = b return [merged[t] for t in sorted(merged.keys())] def update_symbol_history(symbol: str, days_back: int = 7) -> None: """ Aktualisiert eine bestehende History-Datei, indem die letzten Tage neu geladen und gemerged werden. """ api_key = load_api_key() data = load_symbol_history(symbol) today = datetime.now(timezone.utc).date() start_date = today - timedelta(days=days_back) for key in ["daily", "h1", "m15", "m5", "m1"]: cfg = TIMEFRAMES[key] if key == "m1": new_bars = chunked_fetch_1m( symbol, datetime.combine(start_date, datetime.min.time(), tzinfo=timezone.utc), datetime.combine(today, datetime.min.time(), tzinfo=timezone.utc), api_key, ) else: new_bars = fetch_polygon_range( symbol=symbol, multiplier=cfg["multiplier"], timespan=cfg["timespan"], from_date=start_date.isoformat(), to_date=today.isoformat(), api_key=api_key, ) existing = data.get(key, []) data[key] = merge_bars(existing, new_bars) time.sleep(1) data["last_update"] = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") save_symbol_history(symbol, data) def main(): import argparse parser = argparse.ArgumentParser(description="SPY/QQQ Historie initialisieren oder updaten.") parser.add_argument( "--mode", choices=["init", "update"], default="init", help="init = kompletter 2‑Jahres-Download; update = nur letzte Tage aktualisieren", ) parser.add_argument( "--days-back", type=int, default=730, help="Wie viele Tage zurück geladen bzw. neu geladen werden sollen.", ) args = parser.parse_args() if args.mode == "init": for symbol in SYMBOLS: data = fetch_full_history_for_symbol(symbol, days_back=args.days_back) save_symbol_history(symbol, data) else: for symbol in SYMBOLS: update_symbol_history(symbol, days_back=args.days_back) if __name__ == "__main__": main()