Real price data: yfinance re-ingest + Interactive Brokers source

.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

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

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`

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":

hrp/data/ingestion/prices.py

@@ -5,14 +5,15 @@
 """
 
 import argparse
-from datetime import date, timedelta
+from datetime import date
 from typing import Any
 
 import pandas as pd
 from loguru import logger
 
 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'])}")
 
 

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,18 +68,24 @@ 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")
 
         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: