#!/usr/bin/env python3 """ Ingest market history into Aureon's single-file global history database. This is a pragmatic "master dataset" builder: - Your private account history is handled by sync_global_history_db.py - This script adds *public* market history (OHLCV bars, optionally trades) Important reality check: Downloading "every single trade ever recorded" across all global markets is not feasible with typical API plans (data volume, cost, rate limits, licenses). This script is built to scale up safely by: - explicit symbol selection - explicit time bounds - resumable watermarks stored in SQLite """ from __future__ import annotations import argparse import json import os import sqlite3 import sys import time from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple # Make this script runnable from any working directory by ensuring repo imports work. _REPO_ROOT = Path(__file__).resolve().parents[2] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) try: from dotenv import load_dotenv load_dotenv(_REPO_ROOT / ".env") except Exception: pass def _json_dumps_safe(obj: Any) -> str: try: return json.dumps(obj, ensure_ascii=True, separators=(",", ":"), default=str) except Exception: try: return json.dumps({"_unserializable": True, "repr": repr(obj)}, ensure_ascii=True, separators=(",", ":")) except Exception: return "{}" def _parse_dt(value: str) -> datetime: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) def _dt_to_iso_z(dt: datetime) -> str: # CoinAPI is strict: it rejects fractional seconds for time_start/time_end. return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _dt_from_ms(ts_ms: int) -> datetime: return datetime.fromtimestamp(ts_ms / 1000.0, tz=timezone.utc) def _period_to_seconds(period_id: str) -> int: p = str(period_id or "").strip().upper() for suffix, mult in (("SEC", 1), ("MIN", 60), ("HRS", 3600), ("DAY", 86400)): if p.endswith(suffix): num = p[: -len(suffix)] try: return int(num) * mult except Exception: return 60 return 60 def _default_chunk(period_seconds: int) -> timedelta: # Try to keep requests in a reasonable size window. if period_seconds <= 60: # 1m bars: cap at ~3 days return timedelta(days=3) if period_seconds <= 300: # 5m bars: cap at ~14 days return timedelta(days=14) if period_seconds <= 3600: # hourly: cap at ~180 days return timedelta(days=180) # daily+: cap at ~5 years return timedelta(days=365 * 5) def _extract_venue_from_coinapi_symbol_id(symbol_id: str) -> str: sid = str(symbol_id or "") if "_" not in sid: return "" return sid.split("_", 1)[0].upper() def _symbol_from_coinapi_symbol_id(symbol_id: str) -> str: parts = str(symbol_id or "").split("_") if len(parts) >= 4: base = parts[2].upper() quote = parts[3].upper() return f"{base}{quote}" return str(symbol_id or "").upper() def _dedupe_preserve_order(items: Iterable[str]) -> List[str]: seen: set[str] = set() out: List[str] = [] for it in items: if not it: continue if it in seen: continue seen.add(it) out.append(it) return out def _parse_csv(value: str) -> List[str]: return [v.strip() for v in str(value or "").split(",") if v.strip()] def _insert_market_bars(conn: sqlite3.Connection, rows: Sequence[Tuple[Any, ...]]) -> Tuple[int, int]: if not rows: return (0, 0) stmt = """ INSERT OR IGNORE INTO market_bars( provider, venue, symbol_id, symbol, period_id, time_start_ms, time_end_ms, open, high, low, close, volume, trades_count, ingested_at_ms, raw_json ) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ before = conn.total_changes conn.executemany(stmt, rows) inserted = conn.total_changes - before skipped = len(rows) - inserted return inserted, skipped def _insert_market_trades(conn: sqlite3.Connection, rows: Sequence[Tuple[Any, ...]]) -> Tuple[int, int]: if not rows: return (0, 0) stmt = """ INSERT OR IGNORE INTO market_trades( provider, venue, symbol_id, symbol, trade_id, side, size, price, ts_ms, ingested_at_ms, raw_json ) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ before = conn.total_changes conn.executemany(stmt, rows) inserted = conn.total_changes - before skipped = len(rows) - inserted return inserted, skipped def _coinapi_discover_symbol_ids( client: Any, base: str, quote: str, exchange_allowlist: Optional[set[str]], max_symbols: int, ) -> List[str]: base_u = str(base or "").upper() quote_u = str(quote or "").upper() symbol_filter = f"*_{base_u}_{quote_u}" symbol_ids: List[str] = [] try: quotes = client.get_quotes_current(symbol_filter) or [] for q in quotes: if isinstance(q, dict): sid = str(q.get("symbol_id") or "") if sid: symbol_ids.append(sid) except Exception: quotes = [] # Fallback: heavy call, but more reliable if quote-filter returns nothing. if not symbol_ids: try: symbols = client.get_symbols() or [] for s in symbols: if not isinstance(s, dict): continue if str(s.get("asset_id_base") or "").upper() != base_u: continue if str(s.get("asset_id_quote") or "").upper() != quote_u: continue sid = str(s.get("symbol_id") or "") if sid: symbol_ids.append(sid) except Exception: pass symbol_ids = _dedupe_preserve_order(symbol_ids) if exchange_allowlist: symbol_ids = [sid for sid in symbol_ids if _extract_venue_from_coinapi_symbol_id(sid) in exchange_allowlist] # Prefer SPOT markets first. symbol_ids.sort(key=lambda sid: (0 if "_SPOT_" in sid.upper() else 1, sid)) return symbol_ids[: max(1, int(max_symbols or 1))] def _coinapi_index_symbols(conn: sqlite3.Connection, client: Any, limit: int = 0) -> int: """Cache CoinAPI symbol registry into the DB (symbols table).""" from aureon.core import aureon_global_history_db as ghdb # type: ignore symbols = client.get_symbols() or [] if not isinstance(symbols, list): print("[coinapi] symbols endpoint returned non-list") return 0 total = len(symbols) if limit and limit > 0: total = min(total, int(limit)) inserted = 0 for i, s in enumerate(symbols[:total], start=1): if not isinstance(s, dict): continue symbol_id = str(s.get("symbol_id") or "") if not symbol_id: continue venue = str(s.get("exchange_id") or "") or _extract_venue_from_coinapi_symbol_id(symbol_id) symbol_type = str(s.get("symbol_type") or "") base = str(s.get("asset_id_base") or "") quote = str(s.get("asset_id_quote") or "") canonical_symbol = f"{base}{quote}".upper() if base and quote else _symbol_from_coinapi_symbol_id(symbol_id) data_start_raw = ( s.get("data_start") or s.get("data_trade_start") or s.get("data_quote_start") or s.get("data_orderbook_start") ) data_end_raw = ( s.get("data_end") or s.get("data_trade_end") or s.get("data_quote_end") or s.get("data_orderbook_end") ) data_start_ms: Optional[int] = None if data_start_raw: try: data_start_ms = int(_parse_dt(str(data_start_raw)).timestamp() * 1000) except Exception: data_start_ms = None data_end_ms: Optional[int] = None if data_end_raw: try: data_end_ms = int(_parse_dt(str(data_end_raw)).timestamp() * 1000) except Exception: data_end_ms = None ghdb.upsert_symbol( conn, { "provider": "coinapi", "symbol_id": symbol_id, "venue": venue, "symbol": canonical_symbol, "symbol_type": symbol_type, "asset_id_base": base.upper(), "asset_id_quote": quote.upper(), "data_start_ms": data_start_ms, "data_end_ms": data_end_ms, "raw_json": _json_dumps_safe(s), }, ) inserted += 1 if inserted % 5000 == 0: conn.commit() print(f"[coinapi] indexed {inserted}/{total} symbols...") conn.commit() print(f"[coinapi] indexed symbols: {inserted}") return inserted def _coinapi_split_symbol_id(symbol_id: str) -> Tuple[str, str, str]: parts = str(symbol_id or "").split("_") venue = parts[0].upper() if len(parts) >= 1 and parts[0] else "" base = parts[2].upper() if len(parts) >= 4 and parts[2] else "" quote = parts[3].upper() if len(parts) >= 4 and parts[3] else "" return venue, base, quote def _ingest_coinapi_symbol_id( conn: sqlite3.Connection, client: Any, start_dt: datetime, end_dt: datetime, symbol_id: str, venue: str, symbol: str, period_id_u: str, chunk: timedelta, include_trades: bool, no_resume: bool, ) -> None: # local import to keep the script standalone if core isn't installed from aureon.core import aureon_global_history_db as ghdb # type: ignore wm_key = f"bars:coinapi:{symbol_id}:{period_id_u}" wm = None if no_resume else ghdb.get_watermark_ms(conn, wm_key) cur_start = start_dt if wm is not None: cur_start = max(cur_start, _dt_from_ms(int(wm) + 1)) total_inserted = 0 total_skipped = 0 loops = 0 while cur_start < end_dt: loops += 1 cur_end = min(cur_start + chunk, end_dt) bars = client.get_ohlcv_history( symbol_id=symbol_id, period=period_id_u, time_start=_dt_to_iso_z(cur_start), time_end=_dt_to_iso_z(cur_end), limit=10000, ) status = getattr(client, "last_status_code", None) if status != 200: err = str(getattr(client, "last_error_text", "") or "") if status == 429: time.sleep(2.0) continue print(f"[coinapi] HTTP {status} for {symbol_id}: {err[:200]}") break ingested_at_ms = int(time.time() * 1000) rows: List[Tuple[Any, ...]] = [] max_end_ms: Optional[int] = None for b in bars or []: if not isinstance(b, dict): continue ts_start = str(b.get("time_period_start") or "") if not ts_start: continue try: ts_start_ms = int(_parse_dt(ts_start).timestamp() * 1000) except Exception: continue ts_end_ms: Optional[int] = None ts_end = str(b.get("time_period_end") or "") if ts_end: try: ts_end_ms = int(_parse_dt(ts_end).timestamp() * 1000) except Exception: ts_end_ms = None if ts_end_ms is not None: max_end_ms = ts_end_ms if max_end_ms is None else max(max_end_ms, ts_end_ms) else: max_end_ms = ts_start_ms if max_end_ms is None else max(max_end_ms, ts_start_ms) rows.append( ( "coinapi", venue, symbol_id, symbol, period_id_u, ts_start_ms, ts_end_ms, float(b.get("price_open", 0) or 0), float(b.get("price_high", 0) or 0), float(b.get("price_low", 0) or 0), float(b.get("price_close", 0) or 0), float(b.get("volume_traded", 0) or 0), int(b.get("trades_count", 0) or 0), ingested_at_ms, _json_dumps_safe(b), ) ) ins, skip = _insert_market_bars(conn, rows) total_inserted += ins total_skipped += skip # Advance start if max_end_ms is not None: ghdb.set_watermark_ms(conn, wm_key, int(max_end_ms)) conn.commit() cur_start = _dt_from_ms(int(max_end_ms) + 1) else: # No data returned for this window; advance. ghdb.set_watermark_ms(conn, wm_key, int(cur_end.timestamp() * 1000)) conn.commit() cur_start = cur_end # Gentle pacing on free plans. time.sleep(0.11) # Avoid infinite loops if time parsing fails. if loops > 200000: print(f"[coinapi] aborting {symbol_id}: too many loops") break print(f"[coinapi] bars {symbol_id} {period_id_u}: inserted={total_inserted} skipped={total_skipped}") if not include_trades: return trade_wm_key = f"trades:coinapi:{symbol_id}" trade_wm = None if no_resume else ghdb.get_watermark_ms(conn, trade_wm_key) trade_start = start_dt if trade_wm is not None: trade_start = max(trade_start, _dt_from_ms(int(trade_wm) + 1)) # Trades are much heavier; use shorter windows. trade_chunk = timedelta(minutes=15) trades_inserted = 0 trades_skipped = 0 loops = 0 while trade_start < end_dt: loops += 1 trade_end = min(trade_start + trade_chunk, end_dt) trades = client.get_trades_history( symbol_id=symbol_id, time_start=_dt_to_iso_z(trade_start), time_end=_dt_to_iso_z(trade_end), limit=10000, ) status = getattr(client, "last_status_code", None) if status != 200: err = str(getattr(client, "last_error_text", "") or "") if status == 429: time.sleep(2.0) continue print(f"[coinapi] HTTP {status} for trades {symbol_id}: {err[:200]}") break ingested_at_ms = int(time.time() * 1000) t_rows: List[Tuple[Any, ...]] = [] max_ts_ms: Optional[int] = None for t in trades or []: if not isinstance(t, dict): continue trade_id = str(t.get("uuid") or t.get("trade_id") or t.get("id") or "") if not trade_id: continue ts_raw = str(t.get("time_exchange") or t.get("time_coinapi") or "") ts_ms: Optional[int] = None if ts_raw: try: ts_ms = int(_parse_dt(ts_raw).timestamp() * 1000) except Exception: ts_ms = None if ts_ms is not None: max_ts_ms = ts_ms if max_ts_ms is None else max(max_ts_ms, ts_ms) t_rows.append( ( "coinapi", venue, symbol_id, symbol, trade_id, str(t.get("taker_side") or t.get("side") or "").upper(), float(t.get("size", 0) or 0), float(t.get("price", 0) or 0), ts_ms, ingested_at_ms, _json_dumps_safe(t), ) ) ins, skip = _insert_market_trades(conn, t_rows) trades_inserted += ins trades_skipped += skip if max_ts_ms is not None: ghdb.set_watermark_ms(conn, trade_wm_key, int(max_ts_ms)) conn.commit() trade_start = _dt_from_ms(int(max_ts_ms) + 1) else: ghdb.set_watermark_ms(conn, trade_wm_key, int(trade_end.timestamp() * 1000)) conn.commit() trade_start = trade_end time.sleep(0.11) if loops > 200000: print(f"[coinapi] aborting trades {symbol_id}: too many loops") break print(f"[coinapi] trades {symbol_id}: inserted={trades_inserted} skipped={trades_skipped}") def ingest_coinapi_all_symbols( conn: sqlite3.Connection, start_dt: datetime, end_dt: datetime, period_id: str, exchange_allowlist: Optional[set[str]], symbol_type_filter: str, base_assets: List[str], quote_assets: List[str], symbol_limit: int, include_trades: bool, no_resume: bool, ) -> None: """Ingest CoinAPI bars/trades for every symbol_id in the CoinAPI registry.""" try: from aureon.exchanges.coinapi_anomaly_detector import CoinAPIClient # type: ignore except Exception: from coinapi_anomaly_detector import CoinAPIClient # type: ignore api_key = os.getenv("COINAPI_KEY", "") or os.getenv("COINAPI_API_KEY", "") if not api_key: print("[coinapi] COINAPI_KEY missing, skipping.") return client = CoinAPIClient(api_key) period_id_u = str(period_id or "1MIN").upper() period_seconds = _period_to_seconds(period_id_u) chunk = _default_chunk(period_seconds) # Ensure we have the registry locally. try: count_row = conn.execute( "SELECT COUNT(1) AS n FROM symbols WHERE provider = 'coinapi';" ).fetchone() n = int(count_row["n"]) if count_row and "n" in count_row.keys() else int(count_row[0] if count_row else 0) except Exception: n = 0 if n <= 0: print("[coinapi] symbols table empty; indexing CoinAPI registry first (this is a big call).") _coinapi_index_symbols(conn, client, limit=0) rows = conn.execute( "SELECT symbol_id, venue, symbol, symbol_type, asset_id_base, asset_id_quote " "FROM symbols WHERE provider = 'coinapi';" ).fetchall() st_filter = str(symbol_type_filter or "").strip().upper() base_set = {b.strip().upper() for b in base_assets if b.strip()} quote_set = {q.strip().upper() for q in quote_assets if q.strip()} universe: List[Tuple[str, str, str]] = [] for r in rows or []: sid = str(r["symbol_id"] or "") if not sid: continue venue = str(r["venue"] or "") or _extract_venue_from_coinapi_symbol_id(sid) symbol = str(r["symbol"] or "") or _symbol_from_coinapi_symbol_id(sid) sym_type = str(r["symbol_type"] or "").strip().upper() base = str(r["asset_id_base"] or "").strip().upper() quote = str(r["asset_id_quote"] or "").strip().upper() if not base or not quote: _, base2, quote2 = _coinapi_split_symbol_id(sid) base = base or base2 quote = quote or quote2 if exchange_allowlist and venue.upper() not in exchange_allowlist: continue if st_filter and sym_type and sym_type != st_filter: continue if st_filter and not sym_type: # If we asked for a type but the registry row doesn't have one, skip. continue if base_set and base not in base_set: continue if quote_set and quote not in quote_set: continue universe.append((sid, venue.upper(), symbol.upper())) # Stable order and optional limit. universe.sort(key=lambda t: (t[1], t[0])) if symbol_limit and symbol_limit > 0: universe = universe[: int(symbol_limit)] print(f"[coinapi] ingest universe size: {len(universe)}") for sid, venue, symbol in universe: _ingest_coinapi_symbol_id( conn=conn, client=client, start_dt=start_dt, end_dt=end_dt, symbol_id=sid, venue=venue, symbol=symbol, period_id_u=period_id_u, chunk=chunk, include_trades=include_trades, no_resume=no_resume, ) def ingest_coinapi( conn: sqlite3.Connection, start_dt: datetime, end_dt: datetime, pairs: List[str], period_id: str, exchange_allowlist: Optional[set[str]], max_symbols_per_pair: int, include_trades: bool, no_resume: bool, ) -> None: try: from aureon.exchanges.coinapi_anomaly_detector import CoinAPIClient # type: ignore except Exception: from coinapi_anomaly_detector import CoinAPIClient # type: ignore api_key = os.getenv("COINAPI_KEY", "") or os.getenv("COINAPI_API_KEY", "") if not api_key: print("[coinapi] COINAPI_KEY missing, skipping.") return client = CoinAPIClient(api_key) period_id_u = str(period_id or "1MIN").upper() period_seconds = _period_to_seconds(period_id_u) chunk = _default_chunk(period_seconds) for pair in pairs: if "/" not in pair: print(f"[coinapi] skipping invalid pair (expected BASE/QUOTE): {pair}") continue base, quote = pair.split("/", 1) base = base.strip().upper() quote = quote.strip().upper() if not base or not quote: continue symbol_ids = _coinapi_discover_symbol_ids( client=client, base=base, quote=quote, exchange_allowlist=exchange_allowlist, max_symbols=max_symbols_per_pair, ) if not symbol_ids: print(f"[coinapi] no symbol_ids found for {base}/{quote}") continue print(f"[coinapi] {base}/{quote}: {len(symbol_ids)} symbol_ids") for symbol_id in symbol_ids: venue = _extract_venue_from_coinapi_symbol_id(symbol_id) symbol = _symbol_from_coinapi_symbol_id(symbol_id) _ingest_coinapi_symbol_id( conn=conn, client=client, start_dt=start_dt, end_dt=end_dt, symbol_id=symbol_id, venue=venue, symbol=symbol, period_id_u=period_id_u, chunk=chunk, include_trades=include_trades, no_resume=no_resume, ) def ingest_alpaca_stocks( conn: sqlite3.Connection, start_dt: datetime, end_dt: datetime, symbols: List[str], timeframe: str, feed: str, no_resume: bool, ) -> None: import requests api_key = os.getenv("ALPACA_API_KEY", "") or os.getenv("APCA_API_KEY_ID", "") api_secret = os.getenv("ALPACA_SECRET_KEY", "") or os.getenv("ALPACA_API_SECRET") or os.getenv("APCA_API_SECRET_KEY", "") if not api_key or not api_secret: print("[alpaca] ALPACA_API_KEY/ALPACA_SECRET_KEY missing, skipping.") return data_url = "https://data.alpaca.markets" headers = {"APCA-API-KEY-ID": api_key, "APCA-API-SECRET-KEY": api_secret} # local import to keep the script standalone if core isn't installed from aureon.core import aureon_global_history_db as ghdb # type: ignore tf = str(timeframe or "1Min") # Normalize to the CoinAPI-ish "1MIN" style for consistent keys. period_id = tf.replace("Hour", "HRS").replace("Min", "MIN").replace("Day", "DAY").upper() period_seconds = _period_to_seconds(period_id) for sym in symbols: sym_u = str(sym or "").strip().upper() if not sym_u: continue wm_key = f"bars:alpaca:{sym_u}:{period_id}" wm = None if no_resume else ghdb.get_watermark_ms(conn, wm_key) cur_start = start_dt if wm is not None: cur_start = max(cur_start, _dt_from_ms(int(wm) + 1)) inserted_total = 0 skipped_total = 0 page_token: Optional[str] = None while cur_start < end_dt: params: Dict[str, Any] = { "timeframe": tf, "start": _dt_to_iso_z(cur_start), "end": _dt_to_iso_z(end_dt), "limit": 10000, "adjustment": "raw", "feed": str(feed or "iex").lower(), "sort": "asc", } if page_token: params["page_token"] = page_token try: resp = requests.get( f"{data_url}/v2/stocks/{sym_u}/bars", headers=headers, params=params, timeout=15, ) except Exception as exc: print(f"[alpaca] request error {sym_u}: {exc}") break if resp.status_code != 200: print(f"[alpaca] {sym_u} HTTP {resp.status_code}: {resp.text[:200]}") break payload = resp.json() or {} bars = payload.get("bars") or [] page_token = payload.get("next_page_token") or None ingested_at_ms = int(time.time() * 1000) rows: List[Tuple[Any, ...]] = [] max_ts_ms: Optional[int] = None for b in bars: if not isinstance(b, dict): continue ts_raw = str(b.get("t") or "") if not ts_raw: continue try: ts_ms = int(_parse_dt(ts_raw).timestamp() * 1000) except Exception: continue max_ts_ms = ts_ms if max_ts_ms is None else max(max_ts_ms, ts_ms) time_end_ms = ts_ms + int(period_seconds * 1000) rows.append( ( "alpaca", "alpaca", sym_u, sym_u, period_id, ts_ms, time_end_ms, float(b.get("o", 0) or 0), float(b.get("h", 0) or 0), float(b.get("l", 0) or 0), float(b.get("c", 0) or 0), float(b.get("v", 0) or 0), int(b.get("n", 0) or 0), ingested_at_ms, _json_dumps_safe(b), ) ) ins, skip = _insert_market_bars(conn, rows) inserted_total += ins skipped_total += skip if max_ts_ms is not None: ghdb.set_watermark_ms(conn, wm_key, int(max_ts_ms)) conn.commit() cur_start = _dt_from_ms(int(max_ts_ms) + 1) # No more pages or no data: we're caught up to end_dt. if not page_token or not bars: break time.sleep(0.11) print(f"[alpaca] bars {sym_u} {period_id}: inserted={inserted_total} skipped={skipped_total}") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--db", default="", help="Override SQLite path (default: state/aureon_global_history.sqlite)") parser.add_argument("--no-resume", action="store_true", help="Ignore stored watermarks and re-ingest from --start") parser.add_argument("--start", default="", help="ISO start datetime (default: now - --days)") parser.add_argument("--end", default="", help="ISO end datetime (default: now)") parser.add_argument("--days", type=int, default=7, help="If --start not set, use now - DAYS (default: 7)") parser.add_argument("--coinapi", action="store_true", help="Ingest CoinAPI crypto market history") parser.add_argument("--coinapi-index", action="store_true", help="Cache CoinAPI symbol registry into the DB") parser.add_argument("--coinapi-index-limit", type=int, default=0, help="Limit CoinAPI symbols indexed (0=all)") parser.add_argument("--coinapi-all-symbols", action="store_true", help="Ingest CoinAPI history for all symbol_ids in the registry") parser.add_argument("--coinapi-pairs", default="BTC/USD,ETH/USD", help="Comma pairs BASE/QUOTE") parser.add_argument("--coinapi-period", default="1MIN", help="CoinAPI period_id (e.g. 1MIN, 5MIN, 1HRS, 1DAY)") parser.add_argument("--coinapi-exchanges", default="", help="Comma allowlist of exchange_ids (e.g. BINANCE,COINBASE)") parser.add_argument("--coinapi-symbol-type", default="", help="Filter by symbol_type in all-symbols mode (blank=all)") parser.add_argument("--coinapi-base-assets", default="", help="CSV base-asset filter for all-symbols mode (e.g. BTC,ETH)") parser.add_argument("--coinapi-quote-assets", default="", help="CSV quote-asset filter for all-symbols mode (e.g. USD,USDT,USDC)") parser.add_argument("--coinapi-symbol-limit", type=int, default=0, help="Limit symbol_ids ingested in all-symbols mode (0=all)") parser.add_argument("--coinapi-max-symbols", type=int, default=6, help="Max symbol_ids per pair (pair mode)") parser.add_argument("--coinapi-trades", action="store_true", help="Also ingest CoinAPI trade ticks (heavy)") parser.add_argument("--alpaca", action="store_true", help="Ingest Alpaca stock market history") parser.add_argument("--alpaca-stocks", default="AAPL,MSFT,TSLA", help="Comma stock symbols") parser.add_argument("--alpaca-timeframe", default="1Min", help="Alpaca timeframe (e.g. 1Min, 5Min, 1Hour, 1Day)") parser.add_argument("--alpaca-feed", default="iex", help="Alpaca data feed (iex or sip)") args = parser.parse_args() now = datetime.now(tz=timezone.utc) end_dt = _parse_dt(args.end) if args.end else now if args.start: start_dt = _parse_dt(args.start) else: start_dt = end_dt - timedelta(days=max(1, int(args.days or 1))) if start_dt >= end_dt: print("start must be < end") return 2 from aureon.core import aureon_global_history_db as ghdb # type: ignore conn = ghdb.connect(args.db or None) try: allowlist = {s.strip().upper() for s in _parse_csv(args.coinapi_exchanges)} or None if args.coinapi_index: try: from aureon.exchanges.coinapi_anomaly_detector import CoinAPIClient # type: ignore except Exception: from coinapi_anomaly_detector import CoinAPIClient # type: ignore api_key = os.getenv("COINAPI_KEY", "") or os.getenv("COINAPI_API_KEY", "") if not api_key: print("[coinapi] COINAPI_KEY missing, cannot index.") else: client = CoinAPIClient(api_key) _coinapi_index_symbols(conn, client, limit=int(args.coinapi_index_limit or 0)) if args.coinapi_all_symbols: ingest_coinapi_all_symbols( conn=conn, start_dt=start_dt, end_dt=end_dt, period_id=args.coinapi_period, exchange_allowlist=allowlist, symbol_type_filter=str(args.coinapi_symbol_type or ""), base_assets=_parse_csv(args.coinapi_base_assets), quote_assets=_parse_csv(args.coinapi_quote_assets), symbol_limit=int(args.coinapi_symbol_limit or 0), include_trades=bool(args.coinapi_trades), no_resume=bool(args.no_resume), ) elif args.coinapi: pairs = _parse_csv(args.coinapi_pairs) ingest_coinapi( conn=conn, start_dt=start_dt, end_dt=end_dt, pairs=pairs, period_id=args.coinapi_period, exchange_allowlist=allowlist, max_symbols_per_pair=int(args.coinapi_max_symbols or 1), include_trades=bool(args.coinapi_trades), no_resume=bool(args.no_resume), ) if args.alpaca: ingest_alpaca_stocks( conn=conn, start_dt=start_dt, end_dt=end_dt, symbols=_parse_csv(args.alpaca_stocks), timeframe=args.alpaca_timeframe, feed=args.alpaca_feed, no_resume=bool(args.no_resume), ) finally: try: conn.close() except Exception: pass return 0 if __name__ == "__main__": raise SystemExit(main())