diff --git a/.env.example b/.env.example index fb122d9..08c478e 100644 --- a/.env.example +++ b/.env.example @@ -117,6 +117,10 @@ IBKR_PORT=7497 IBKR_CLIENT_ID=1 IBKR_ACCOUNT=DU123456 IBKR_PAPER_TRADING=true +# IBKR as a historical price source (needs IB Gateway/TWS running + ib-insync). +# Use a client id distinct from trading; pacing pause between symbols (seconds). +IBKR_DATA_CLIENT_ID=11 +HRP_IBKR_PACE_SECONDS=11 # ----------------------------------------------------------------------------- # Broker Selection diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fd9c54..41397d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ version is auto-bumped by `scripts/bump_version.py`; it is never hand-set. ## [Unreleased] +## [2026.629.1] - 2026-06-29 + +### Added +- Interactive Brokers historical price source (`ibkr`), selectable via + `run_job --price-source ibkr`; yfinance remains the default. + +### Changed +- Replaced synthetic price data with real yfinance history (prices + features + re-ingested for the full universe). + ## [2026.629.0] - 2026-06-29 ## [2026.628.0] - 2026-06-28 diff --git a/docs/operations/running-the-platform.md b/docs/operations/running-the-platform.md index 766ab75..db49810 100644 --- a/docs/operations/running-the-platform.md +++ b/docs/operations/running-the-platform.md @@ -94,6 +94,20 @@ To backfill a specific date range, see [Data Backfill](data-backfill.md). Schedule these automatically via `hrp start --full` or the launchd jobs (`./scripts/manage_launchd.sh install`). +**Data source.** Prices default to **yfinance** (free, no key — good for the full +historical backfill). For broker-grade daily data from **Interactive Brokers** +(needs IB Gateway/TWS running + the trading extra), select it per run: + +```bash +pip install -e ".[trading]" # ib-insync +python -m hrp.agents.run_job --job prices --price-source ibkr +``` + +IBKR paces historical requests, so it's best for ongoing/daily updates rather +than a multi-year first backfill (use yfinance for that). Falls back to yfinance +if the Gateway isn't reachable. **Robinhood is execution-only** — not a price +source. + ## 5. Configure API keys (`.env`) Most of the platform runs without keys, but some features need them. Edit `.env` diff --git a/hrp/agents/run_job.py b/hrp/agents/run_job.py index 2042dc2..3f3a097 100644 --- a/hrp/agents/run_job.py +++ b/hrp/agents/run_job.py @@ -28,7 +28,6 @@ from loguru import logger - # Configure logging to file + stderr LOG_DIR = "~/hrp-data/logs" @@ -51,15 +50,15 @@ def _setup_logging(job_name: str) -> None: ) -def run_prices(dry_run: bool = False) -> dict: - """Run daily price ingestion.""" +def run_prices(dry_run: bool = False, source: str = "yfinance") -> dict: + """Run daily price ingestion. source: yfinance | polygon | ibkr.""" from hrp.agents.jobs import PriceIngestionJob if dry_run: - logger.info("[DRY RUN] Would run price ingestion") + logger.info(f"[DRY RUN] Would run price ingestion (source={source})") return {"status": "dry_run", "job": "prices"} - job = PriceIngestionJob(symbols=None) + job = PriceIngestionJob(symbols=None, source=source) return job.run() @@ -136,9 +135,7 @@ def run_fundamentals_backfill(dry_run: bool = False, days: int = 365) -> dict: return job.run() -def run_signal_scan( - dry_run: bool = False, ic_threshold: float = 0.03 -) -> dict: +def run_signal_scan(dry_run: bool = False, ic_threshold: float = 0.03) -> dict: """Run weekly signal scan.""" from hrp.agents.research_agents import SignalScientist @@ -282,6 +279,7 @@ def on_validation_analyst_complete(event: dict) -> None: if passed > 0: logger.info(f"Triggering Risk Manager for {passed} passed hypotheses") from hrp.agents.risk_manager import RiskManager + risk_mgr = RiskManager(hypothesis_ids=None, send_alerts=True) risk_mgr.run() @@ -298,6 +296,7 @@ def on_risk_manager_assessment(event: dict) -> None: if passed > 0: logger.info(f"Triggering CIO Agent for {passed} risk-cleared hypotheses") from hrp.agents.cio import CIOAgent + agent = CIOAgent( job_id=f"cio-triggered-{date.today().strftime('%Y%m%d')}", actor="agent:cio", @@ -454,7 +453,7 @@ def run_drift_monitor(dry_run: bool = False, auto_rollback: bool = False) -> dic dry_run: If True, skip job entirely auto_rollback: If True, automatically rollback drifting models """ - from hrp.agents.drift_monitor_job import DriftMonitorJob, DriftConfig + from hrp.agents.drift_monitor_job import DriftConfig, DriftMonitorJob if dry_run: logger.info("[DRY RUN] Would run drift monitor job") @@ -539,6 +538,13 @@ def main() -> None: choices=["simfin", "yfinance"], help="Data source for fundamentals job (default: simfin)", ) + parser.add_argument( + "--price-source", + type=str, + default="yfinance", + choices=["yfinance", "polygon", "ibkr"], + help="Data source for the prices job (default: yfinance)", + ) parser.add_argument( "--days", type=int, @@ -568,6 +574,8 @@ def main() -> None: kwargs: dict = {"dry_run": args.dry_run} if args.job == "signal-scan": kwargs["ic_threshold"] = args.ic_threshold + elif args.job == "prices": + kwargs["source"] = args.price_source elif args.job == "fundamentals": kwargs["source"] = args.fundamentals_source elif args.job == "fundamentals-backfill": diff --git a/hrp/data/ingestion/prices.py b/hrp/data/ingestion/prices.py index 649ca97..478f967 100644 --- a/hrp/data/ingestion/prices.py +++ b/hrp/data/ingestion/prices.py @@ -5,7 +5,7 @@ """ import argparse -from datetime import date, timedelta +from datetime import date from typing import Any import pandas as pd @@ -13,6 +13,7 @@ from hrp.data.constants import TEST_SYMBOLS from hrp.data.db import get_db +from hrp.data.sources.base import DataSourceBase from hrp.data.sources.polygon_source import PolygonSource from hrp.data.sources.yfinance_source import YFinanceSource @@ -45,8 +46,8 @@ def ingest_prices( db = get_db() # Initialize primary data source with fallback - primary_source: PolygonSource | YFinanceSource | None = None - fallback_source: PolygonSource | YFinanceSource | None = None + primary_source: DataSourceBase | None = None + fallback_source: DataSourceBase | None = None if source == "polygon": try: @@ -62,8 +63,19 @@ def ingest_prices( primary_source = YFinanceSource() fallback_source = None logger.info("Using YFinance as primary source") + elif source == "ibkr": + try: + from hrp.data.sources.ibkr_source import IBKRDataSource + + primary_source = IBKRDataSource() + fallback_source = YFinanceSource() + logger.info("Using IBKR as primary source with YFinance fallback") + except Exception as e: + logger.warning(f"IBKR unavailable ({e}), falling back to YFinance") + primary_source = YFinanceSource() + fallback_source = None else: - raise ValueError(f"Unknown source: {source}. Use 'polygon' or 'yfinance'") + raise ValueError(f"Unknown source: {source}. Use 'polygon', 'yfinance', or 'ibkr'") stats: dict[str, Any] = { "symbols_requested": len(symbols), @@ -80,7 +92,9 @@ def ingest_prices( used_fallback = False try: - logger.info(f"Fetching {symbol} from {start} to {end} using {primary_source.source_name}") + logger.info( + f"Fetching {symbol} from {start} to {end} using {primary_source.source_name}" + ) # Try primary source df = primary_source.get_daily_bars(symbol, start, end) @@ -121,7 +135,9 @@ def ingest_prices( stats["rows_inserted"] += rows_inserted stats["symbols_success"] += 1 - source_used = fallback_source.source_name if used_fallback else primary_source.source_name + source_used = ( + fallback_source.source_name if used_fallback else primary_source.source_name + ) logger.info(f"Inserted {rows_inserted} rows for {symbol} from {source_used}") except Exception as e: @@ -149,7 +165,7 @@ def _upsert_prices(db, df: pd.DataFrame) -> int: return 0 # Prepare data for insertion - records = df.to_dict('records') + records = df.to_dict("records") with db.connection() as conn: # Create temporary table for bulk insert @@ -158,20 +174,23 @@ def _upsert_prices(db, df: pd.DataFrame) -> int: # Insert into temp table for record in records: - conn.execute(""" + conn.execute( + """ INSERT INTO temp_prices (symbol, date, open, high, low, close, adj_close, volume, source) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, ( - record['symbol'], - record['date'], - record.get('open'), - record.get('high'), - record.get('low'), - record['close'], - record.get('adj_close'), - record.get('volume'), - record.get('source', 'unknown'), - )) + """, + ( + record["symbol"], + record["date"], + record.get("open"), + record.get("high"), + record.get("low"), + record["close"], + record.get("adj_close"), + record.get("volume"), + record.get("source", "unknown"), + ), + ) # Upsert from temp to main table conn.execute(""" @@ -198,9 +217,7 @@ def get_price_stats() -> dict[str, Any]: symbols = conn.execute("SELECT COUNT(DISTINCT symbol) FROM prices").fetchone()[0] # Date range - date_range = conn.execute( - "SELECT MIN(date), MAX(date) FROM prices" - ).fetchone() + date_range = conn.execute("SELECT MIN(date), MAX(date) FROM prices").fetchone() # Rows per symbol per_symbol = conn.execute(""" @@ -218,8 +235,7 @@ def get_price_stats() -> dict[str, Any]: "end": date_range[1], }, "per_symbol": [ - {"symbol": r[0], "rows": r[1], "start": r[2], "end": r[3]} - for r in per_symbol + {"symbol": r[0], "rows": r[1], "start": r[2], "end": r[3]} for r in per_symbol ], } @@ -263,12 +279,12 @@ def main(): if args.stats: stats = get_price_stats() - print(f"\nPrice Data Statistics:") + print("\nPrice Data Statistics:") print(f" Total rows: {stats['total_rows']:,}") print(f" Unique symbols: {stats['unique_symbols']}") print(f" Date range: {stats['date_range']['start']} to {stats['date_range']['end']}") - print(f"\nPer Symbol:") - for s in stats['per_symbol']: + print("\nPer Symbol:") + for s in stats["per_symbol"]: print(f" {s['symbol']:6} {s['rows']:6,} rows ({s['start']} to {s['end']})") return @@ -282,10 +298,10 @@ def main(): source=args.source, ) - print(f"\nIngestion Complete:") + print("\nIngestion Complete:") print(f" Symbols: {stats['symbols_success']}/{stats['symbols_requested']} success") print(f" Rows: {stats['rows_fetched']} fetched, {stats['rows_inserted']} inserted") - if stats['failed_symbols']: + if stats["failed_symbols"]: print(f" Failed: {', '.join(stats['failed_symbols'])}") diff --git a/hrp/data/sources/factory.py b/hrp/data/sources/factory.py index 62e06af..71ed104 100644 --- a/hrp/data/sources/factory.py +++ b/hrp/data/sources/factory.py @@ -3,6 +3,7 @@ from loguru import logger from hrp.data.sources.base import DataSourceBase +from hrp.data.sources.ibkr_source import IBKRDataSource from hrp.data.sources.polygon_source import PolygonSource from hrp.data.sources.yfinance_source import YFinanceSource @@ -26,10 +27,13 @@ class DataSourceFactory: _sources = { "polygon": (PolygonSource, YFinanceSource), "yfinance": (YFinanceSource, None), + "ibkr": (IBKRDataSource, YFinanceSource), } @staticmethod - def create(source: str, with_fallback: bool = True) -> tuple[DataSourceBase, DataSourceBase | None]: + def create( + source: str, with_fallback: bool = True + ) -> tuple[DataSourceBase, DataSourceBase | None]: """ Create data source with optional fallback. @@ -64,10 +68,14 @@ def create(source: str, with_fallback: bool = True) -> tuple[DataSourceBase, Dat if with_fallback and fallback_cls is not None: try: fallback = fallback_cls() - logger.info(f"Using {primary.source_name} as primary with {fallback.source_name} fallback") + logger.info( + f"Using {primary.source_name} as primary with {fallback.source_name} fallback" + ) except ValueError: # Fallback initialization failed - this is unusual but shouldn't prevent primary use - logger.warning(f"{fallback_cls.__name__} initialization failed, using {primary.source_name} only") + logger.warning( + f"{fallback_cls.__name__} initialization failed, using {primary.source_name} only" + ) fallback = None else: logger.info(f"Using {primary.source_name} as primary source") @@ -75,7 +83,9 @@ def create(source: str, with_fallback: bool = True) -> tuple[DataSourceBase, Dat except ValueError as e: # Primary source initialization failed if with_fallback and fallback_cls is not None: - logger.warning(f"{primary_cls.__name__} unavailable ({e}), falling back to {fallback_cls.__name__}") + logger.warning( + f"{primary_cls.__name__} unavailable ({e}), falling back to {fallback_cls.__name__}" + ) primary = fallback_cls() fallback = None else: diff --git a/hrp/data/sources/ibkr_source.py b/hrp/data/sources/ibkr_source.py new file mode 100644 index 0000000..1269c11 --- /dev/null +++ b/hrp/data/sources/ibkr_source.py @@ -0,0 +1,153 @@ +"""Interactive Brokers historical-data adapter. + +Pulls daily OHLCV bars from IBKR via ``ib_insync`` (``reqHistoricalData``). Best +suited to ongoing/daily updates: IBKR paces historical requests (~60 / 10 min), +so a full-universe backfill is slow — one request per symbol with a configurable +pause. Requires IB Gateway / TWS running and logged in, and the optional +``ib-insync`` dependency (``pip install -e ".[trading]"``). + +``ib_insync`` is imported lazily inside the methods, so importing this module +(and registering it in the factory) does not require the package to be installed. +""" + +from __future__ import annotations + +import math +import os +import time +from datetime import date, datetime +from typing import Any + +import pandas as pd +from loguru import logger + +from hrp.data.sources.base import DataSourceBase + +_COLUMNS = ["symbol", "date", "open", "high", "low", "close", "adj_close", "volume", "source"] + + +def _bars_to_df(symbol: str, bars: list[Any], source_name: str) -> pd.DataFrame: + """Map ib_insync historical bars to the standard schema (pure, testable). + + IBKR ``TRADES`` bars are not back-adjusted, so ``adj_close`` mirrors ``close``. + """ + rows = [] + for b in bars: + bar_date = getattr(b, "date", None) + if isinstance(bar_date, datetime): + bar_date = bar_date.date() + close = float(getattr(b, "close")) + rows.append( + { + "symbol": symbol, + "date": bar_date, + "open": float(getattr(b, "open")), + "high": float(getattr(b, "high")), + "low": float(getattr(b, "low")), + "close": close, + "adj_close": close, # IBKR TRADES bars are unadjusted + "volume": int(getattr(b, "volume", 0) or 0), + "source": source_name, + } + ) + if not rows: + return pd.DataFrame() + return pd.DataFrame(rows)[_COLUMNS] + + +def _duration_str(start: date, end: date) -> str: + """IBKR durationStr covering [start, end] for daily bars.""" + days = max(1, (end - start).days) + if days <= 365: + return f"{days} D" + return f"{math.ceil(days / 365)} Y" + + +class IBKRDataSource(DataSourceBase): + """Daily bars from Interactive Brokers (requires IB Gateway/TWS + ib-insync).""" + + source_name = "ibkr" + + def __init__( + self, + host: str | None = None, + port: int | None = None, + client_id: int | None = None, + pace_seconds: float | None = None, + ib: Any | None = None, + ): + super().__init__() + self.host = host or os.getenv("IBKR_HOST", "127.0.0.1") + self.port = int(port or os.getenv("IBKR_PORT", "7497")) + # Distinct client id from the trading connection so both can run. + self.client_id = int(client_id or os.getenv("IBKR_DATA_CLIENT_ID", "11")) + self.pace_seconds = float( + pace_seconds if pace_seconds is not None else os.getenv("HRP_IBKR_PACE_SECONDS", "11") + ) + self._ib = ib # injectable for tests + logger.info("IBKR data source initialized") + + # -- connection --------------------------------------------------------- + def _connect(self) -> None: + if self._ib is not None and self._ib.isConnected(): + return + from ib_insync import IB + + if self._ib is None: + self._ib = IB() + self._ib.connect(self.host, self.port, clientId=self.client_id, timeout=15) + + def _disconnect(self) -> None: + if self._ib is not None and self._ib.isConnected(): + self._ib.disconnect() + + # -- data --------------------------------------------------------------- + def get_daily_bars(self, symbol: str, start: date, end: date) -> pd.DataFrame: + from ib_insync import Stock + + self._connect() + contract = Stock(symbol, "SMART", "USD") + end_dt = datetime(end.year, end.month, end.day, 23, 59, 59) + bars = self._ib.reqHistoricalData( + contract, + endDateTime=end_dt, + durationStr=_duration_str(start, end), + barSizeSetting="1 day", + whatToShow="TRADES", + useRTH=True, + formatDate=1, + ) + if not bars: + logger.warning(f"No IBKR data for {symbol} {start}..{end}") + return pd.DataFrame() + df = _bars_to_df(symbol, bars, self.source_name) + # IBKR returns the whole duration ending at end_dt; clip to [start, end]. + if not df.empty: + df = df[(df["date"] >= start) & (df["date"] <= end)].reset_index(drop=True) + return df + + def get_multiple_symbols(self, symbols: list[str], start: date, end: date) -> pd.DataFrame: + self._connect() + frames = [] + try: + for i, symbol in enumerate(symbols): + try: + df = self.get_daily_bars(symbol, start, end) + if not df.empty: + frames.append(df) + except Exception as exc: + logger.warning(f"Skipping {symbol}: {exc}") + if i < len(symbols) - 1 and self.pace_seconds > 0: + time.sleep(self.pace_seconds) # respect IBKR pacing limits + finally: + self._disconnect() + return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame() + + def validate_symbol(self, symbol: str) -> bool: + try: + from datetime import timedelta + + today = date.today() + return not self.get_daily_bars(symbol, today - timedelta(days=7), today).empty + except Exception: + return False diff --git a/pyproject.toml b/pyproject.toml index 1ea38a7..7326a55 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "hrp" -version = "2026.629.0" +version = "2026.629.1" description = "Hedgefund Research Platform - Personal quantitative research for systematic trading" readme = "README.md" requires-python = ">=3.11" diff --git a/tests/test_data/test_ibkr_source.py b/tests/test_data/test_ibkr_source.py new file mode 100644 index 0000000..6bdd51b --- /dev/null +++ b/tests/test_data/test_ibkr_source.py @@ -0,0 +1,124 @@ +"""Tests for the IBKR historical-data adapter. + +ib_insync is an optional dep and a live Gateway isn't available in CI, so these +mock the ib client (injected) and stub the `ib_insync` module for the parts that +import it. The live reqHistoricalData round-trip is exercised manually against a +running IB Gateway. +""" + +from __future__ import annotations + +import sys +import types +from datetime import date + +import pytest + +from hrp.data.sources.factory import DataSourceFactory +from hrp.data.sources.ibkr_source import ( + IBKRDataSource, + _bars_to_df, + _duration_str, +) + + +class FakeBar: + def __init__(self, d, o, h, low, c, v): + self.date = d + self.open = o + self.high = h + self.low = low + self.close = c + self.volume = v + + +def test_bars_to_df_maps_schema(): + bars = [ + FakeBar(date(2026, 6, 25), 10.0, 11.0, 9.5, 10.5, 1000), + FakeBar(date(2026, 6, 26), 10.5, 12.0, 10.0, 11.8, 2000), + ] + df = _bars_to_df("NVDA", bars, "ibkr") + assert list(df.columns) == [ + "symbol", + "date", + "open", + "high", + "low", + "close", + "adj_close", + "volume", + "source", + ] + assert df.iloc[0]["symbol"] == "NVDA" + assert df.iloc[0]["source"] == "ibkr" + # IBKR TRADES bars are unadjusted -> adj_close mirrors close + assert df.iloc[1]["adj_close"] == 11.8 == df.iloc[1]["close"] + assert df.iloc[1]["volume"] == 2000 + + +def test_bars_to_df_empty(): + assert _bars_to_df("X", [], "ibkr").empty + + +def test_duration_str(): + assert _duration_str(date(2026, 6, 1), date(2026, 6, 20)) == "19 D" + assert _duration_str(date(2024, 6, 1), date(2026, 6, 1)) == "2 Y" + + +def test_factory_registers_ibkr(): + primary, fallback = DataSourceFactory.create("ibkr") + assert isinstance(primary, IBKRDataSource) + assert primary.source_name == "ibkr" + # falls back to yfinance when IBKR is unavailable + assert fallback is not None and fallback.source_name == "yfinance" + + +class FakeIB: + def __init__(self, bars): + self._bars = bars + self.connected = True + self.disconnected = False + + def isConnected(self): + return self.connected + + def connect(self, *a, **k): + self.connected = True + + def disconnect(self): + self.disconnected = True + self.connected = False + + def reqHistoricalData(self, *a, **k): + return self._bars + + +@pytest.fixture +def fake_ib_insync(monkeypatch): + mod = types.ModuleType("ib_insync") + mod.Stock = lambda *a, **k: ("Stock", a) + mod.IB = object + monkeypatch.setitem(sys.modules, "ib_insync", mod) + return mod + + +def test_get_daily_bars_with_injected_client(fake_ib_insync): + bars = [ + FakeBar(date(2026, 6, 24), 1, 2, 1, 1.5, 10), + FakeBar(date(2026, 6, 25), 1, 2, 1, 1.6, 20), + FakeBar(date(2026, 6, 26), 1, 2, 1, 1.7, 30), + ] + src = IBKRDataSource(ib=FakeIB(bars), pace_seconds=0) + df = src.get_daily_bars("AAPL", date(2026, 6, 25), date(2026, 6, 26)) + # clipped to [start, end] + assert list(df["date"]) == [date(2026, 6, 25), date(2026, 6, 26)] + assert df.iloc[0]["close"] == 1.6 + + +def test_get_multiple_symbols_concats_and_disconnects(fake_ib_insync): + bars = [FakeBar(date(2026, 6, 26), 1, 2, 1, 1.7, 30)] + fake = FakeIB(bars) + src = IBKRDataSource(ib=fake, pace_seconds=0) + df = src.get_multiple_symbols(["AAPL", "MSFT"], date(2026, 6, 26), date(2026, 6, 26)) + assert set(df["symbol"]) == {"AAPL", "MSFT"} + assert fake.disconnected is True # connection cleaned up diff --git a/web/package.json b/web/package.json index e41a645..da30748 100644 --- a/web/package.json +++ b/web/package.json @@ -1,6 +1,6 @@ { "name": "web", - "version": "2026.629.0", + "version": "2026.629.1", "private": true, "scripts": { "dev": "next dev",