From e2eb5dd574df883235529cf6e7fab1353120b23c Mon Sep 17 00:00:00 2001 From: Jarno Date: Sun, 26 Jul 2026 14:18:07 +0300 Subject: [PATCH] Data fetcher implementation --- .gitignore | 2 + docs/data-fetcher.md | 40 +++- docs/manual-test/README.md | 46 +++-- src/trading_bot/data/fetch_ibkr_daily.py | 243 +++++++++++++++++++---- 4 files changed, 273 insertions(+), 58 deletions(-) diff --git a/.gitignore b/.gitignore index f02b17f..3e1ca37 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,5 @@ __pycache__/ dist/ build/ *.egg-info/ + +data/ibkr/ diff --git a/docs/data-fetcher.md b/docs/data-fetcher.md index d864696..e937d65 100644 --- a/docs/data-fetcher.md +++ b/docs/data-fetcher.md @@ -12,44 +12,63 @@ The intended example workflow is: - fields: open, high, low, close, volume; - output: one Parquet file named for the ticker, such as `SPY.parquet`. -## Current Skeleton +## Current Implementation The first implementation is intentionally small: -- hard-coded ticker: `SPY`; -- hard-coded range: `2026-06-01` to `2026-06-05`; +- ticker passed as a required command line argument; +- end date passed with `--end-date YYYYMMDD`, defaulting to yesterday; +- end date can also be derived with `--end-date-from-parquet`; +- duration passed with `--duration`, defaulting to `1 W`; - hard-coded IBKR Gateway target: `127.0.0.1:4002`; - uses the IBKR API through `ib_insync`; - prints fetched candles as CSV-like rows; -- does not write Parquet yet; -- does not expose CLI arguments yet; -- does not define storage paths yet. +- writes candles to a symbol-named Parquet file; +- appends to an existing symbol file and keeps one row per date. Run it with: ```sh -mise exec -- uv run python src/trading_bot/data/fetch_ibkr_daily.py +mise exec -- uv run python src/trading_bot/data/fetch_ibkr_daily.py SPY ``` +To override the requested range: + +```sh +mise exec -- uv run python src/trading_bot/data/fetch_ibkr_daily.py SPY --end-date 20250605 --duration "1 M" +``` + +To fetch backward from the oldest date already stored in the symbol file: + +```sh +mise exec -- uv run python src/trading_bot/data/fetch_ibkr_daily.py SPY --end-date-from-parquet +``` + +`--end-date` and `--end-date-from-parquet` cannot be used together. If the symbol Parquet file does not exist or has no rows, `--end-date-from-parquet` uses today's US/Eastern date. + This expects a local IBKR Gateway session to be running and accepting API connections on `127.0.0.1:4002`. +By default, output is written to `data/ibkr/daily/SPY.parquet`. Use `--output-dir` to choose another directory. + Manual test instructions are in [manual-test/README.md](manual-test/README.md). ## Intended Future Behavior -Later, this tool should accept a ticker and date range, fetch daily candles from IBKR, normalize the schema, and write the result to a ticker-named Parquet file. +Later, this tool should broaden configuration around data source and normalization choices. Decided storage behavior: - Parquet files are partitioned by ticker, not by date. - Each ticker should have its own Parquet file, such as `SPY.parquet`. +- The trading date is used as the row key for merges. +- Re-fetching a date replaces the existing row for that date in that symbol's file. Candidate output schema: | Column | Type | Description | | --- | --- | --- | -| `date` | date | Trading session date | -| `ticker` | string | Asset ticker | +| `date` | date index | Trading session date | +| `symbol` | string | Asset ticker | | `open` | float | Daily open price | | `high` | float | Daily high price | | `low` | float | Daily low price | @@ -58,7 +77,6 @@ Candidate output schema: Open decisions: -- where raw and normalized data files should live; - how to handle adjusted versus unadjusted prices; - how to handle missing sessions and IBKR pacing limits; - whether to use `ib_insync` long term or a lower-level IBKR client wrapper. diff --git a/docs/manual-test/README.md b/docs/manual-test/README.md index da172be..20f35d9 100644 --- a/docs/manual-test/README.md +++ b/docs/manual-test/README.md @@ -1,19 +1,21 @@ # Manual Test Instructions -## IBKR Daily Fetcher Skeleton +## IBKR Daily Fetcher -This test checks the first hard-coded IBKR data fetcher without writing any data files. +This test checks the IBKR data fetcher and confirms it writes a symbol-named Parquet file. The fetcher currently requests: - gateway: `127.0.0.1:4002`; - client id: `101`; -- ticker: `SPY`; -- date range: `2026-06-01` to `2026-06-05`; +- symbol: provided as a command line argument; +- end date: provided with `--end-date YYYYMMDD`, defaulting to yesterday; +- end date can also be derived with `--end-date-from-parquet`; +- duration: provided with `--duration`, defaulting to `1 W`; - bar size: `1 day`; - data type: `TRADES`; - regular trading hours only; -- output: printed CSV-like rows. +- output: printed CSV-like rows and a Parquet file in `data/ibkr/daily/`. ## Prerequisites @@ -33,32 +35,52 @@ mise exec -- uv sync From the repository root, run: ```sh -mise exec -- uv run python src/trading_bot/data/fetch_ibkr_daily.py +mise exec -- uv run python src/trading_bot/data/fetch_ibkr_daily.py SPY ``` +To fetch a specific IBKR range, pass an end date and duration: + +```sh +mise exec -- uv run python src/trading_bot/data/fetch_ibkr_daily.py SPY --end-date 20250605 --duration "1 M" +``` + +To fetch backward from the oldest date already stored in `data/ibkr/daily/SPY.parquet`: + +```sh +mise exec -- uv run python src/trading_bot/data/fetch_ibkr_daily.py SPY --end-date-from-parquet +``` + +`--end-date` and `--end-date-from-parquet` cannot be used together. If the symbol Parquet file does not exist or has no rows, `--end-date-from-parquet` uses today's US/Eastern date. + ## Expected Output -The tool should first print the hard-coded request range and connection target: +The tool should first print the request range and connection target: ```text -Fetching SPY daily candles from 2026-06-01 to 2026-06-05 +Fetching SPY daily candles ending 2025-06-05 for duration 1 M Connecting to IBKR Gateway at 127.0.0.1:4002 with client id 101 ``` If the request succeeds, it should then print a header and one row per returned trading day: ```text -date,ticker,open,high,low,close,volume -2026-06-01,SPY,... +date,symbol,open,high,low,close,volume +2025-06-05,SPY,... ``` Exact prices and volume depend on what IBKR returns. +The tool should then write or update: + +```text +data/ibkr/daily/SPY.parquet +``` + +If the Parquet file already exists, rows from the latest fetch are merged into it. The trading date is used as the row key, so a symbol file keeps only one row for each date. + ## Common Issues - Connection refused: IBKR Gateway is not running, the port is not `4002`, or API access is disabled. - Client id already in use: change `IBKR_CLIENT_ID` in the fetcher or disconnect the other client. - No historical bars: confirm the account has market data permissions and that IBKR accepts the requested historical data range. - Pacing or permission errors: note the IBKR error message before changing the request. - -This skeleton does not write Parquet files yet. diff --git a/src/trading_bot/data/fetch_ibkr_daily.py b/src/trading_bot/data/fetch_ibkr_daily.py index 7afed04..b5ef94e 100644 --- a/src/trading_bot/data/fetch_ibkr_daily.py +++ b/src/trading_bot/data/fetch_ibkr_daily.py @@ -1,18 +1,20 @@ -"""Skeleton IBKR daily candle fetcher. - -This intentionally uses hard-coded values while the project shape is still -being designed. Parquet persistence and configurable date ranges come later. -""" +"""IBKR daily candle fetcher.""" from __future__ import annotations +import argparse +import re from dataclasses import dataclass -from datetime import date +from datetime import date, datetime, timedelta +from pathlib import Path +from zoneinfo import ZoneInfo +import pandas as pd -TICKER = "SPY" -START_DATE = date(2026, 6, 1) -END_DATE = date(2026, 6, 5) +DEFAULT_OUTPUT_DIR = Path("data/ibkr/daily") +DEFAULT_DURATION = "1 W" +EASTERN_TZ = ZoneInfo("America/New_York") +DURATION_PATTERN = re.compile(r"^\d+\s+[SDWMY]$") IBKR_HOST = "127.0.0.1" IBKR_PORT = 4002 @@ -32,15 +34,67 @@ class DailyCandle: volume: int -def fetch_hard_coded_daily_candles() -> list[DailyCandle]: - """Fetch one hard-coded week of SPY daily candles from IBKR.""" +def default_end_date() -> date: + """Return yesterday's date in the US/Eastern market timezone.""" + + return datetime.now(EASTERN_TZ).date() - timedelta(days=1) + + +def current_market_date() -> date: + """Return today's date in the US/Eastern market timezone.""" + + return datetime.now(EASTERN_TZ).date() + + +def parse_end_date(value: str) -> date: + """Parse an end date in IBKR-friendly YYYYMMDD form.""" + + try: + return datetime.strptime(value, "%Y%m%d").date() + except ValueError as exc: + raise argparse.ArgumentTypeError( + "end date must use YYYYMMDD format, such as 20250605" + ) from exc + + +def parse_duration(value: str) -> str: + """Normalize and validate an IBKR duration string like '1 W' or '1 M'.""" + + normalized = " ".join(value.upper().split()) + if not DURATION_PATTERN.fullmatch(normalized): + raise argparse.ArgumentTypeError( + "duration must look like an IBKR value, such as '1 W' or '1 M'" + ) + return normalized + + +def format_ibkr_end_datetime(end_date: date) -> str: + """Convert a date to the US/Eastern end datetime string IBKR expects.""" + + return f"{end_date:%Y%m%d} 23:59:59 US/Eastern" + + +def normalize_bar_date(value: date | datetime | str) -> date: + """Normalize an IBKR historical bar date value.""" + + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + return parse_end_date(value) + + +def fetch_daily_candles( + symbol: str, end_date: date, duration: str +) -> list[DailyCandle]: + """Fetch daily candles for a symbol from IBKR.""" try: from ib_insync import IB, Stock # type: ignore[import-not-found] except ImportError as exc: raise SystemExit( "Missing dependency: ib_insync. Install it before running the " - "IBKR fetcher skeleton." + "IBKR fetcher." ) from exc ib = IB() @@ -56,12 +110,12 @@ def fetch_hard_coded_daily_candles() -> list[DailyCandle]: timeout=IBKR_CONNECT_TIMEOUT_SECONDS, ) - contract = Stock(TICKER, "SMART", "USD") + contract = Stock(symbol, "SMART", "USD") ib.qualifyContracts(contract) bars = ib.reqHistoricalData( contract, - endDateTime="20260605 23:59:59 US/Eastern", - durationStr="1 W", + endDateTime=format_ibkr_end_datetime(end_date), + durationStr=duration, barSizeSetting="1 day", whatToShow="TRADES", useRTH=True, @@ -69,22 +123,21 @@ def fetch_hard_coded_daily_candles() -> list[DailyCandle]: ) if not bars: - print("IBKR returned no historical bars for the hard-coded request.") + print(f"IBKR returned no historical bars for {symbol}.") candles: list[DailyCandle] = [] for bar in bars: - trading_day = bar.date if isinstance(bar.date, date) else bar.date.date() - if START_DATE <= trading_day <= END_DATE: - candles.append( - DailyCandle( - trading_day=trading_day, - open=float(bar.open), - high=float(bar.high), - low=float(bar.low), - close=float(bar.close), - volume=int(bar.volume), - ) + trading_day = normalize_bar_date(bar.date) + candles.append( + DailyCandle( + trading_day=trading_day, + open=float(bar.open), + high=float(bar.high), + low=float(bar.low), + close=float(bar.close), + volume=int(bar.volume), ) + ) return candles finally: @@ -92,14 +145,83 @@ def fetch_hard_coded_daily_candles() -> list[DailyCandle]: ib.disconnect() -def print_candles(candles: list[DailyCandle]) -> None: +def candles_to_frame(symbol: str, candles: list[DailyCandle]) -> pd.DataFrame: + """Convert candles to a date-indexed dataframe ready for Parquet storage.""" + + rows = [ + { + "date": candle.trading_day, + "symbol": symbol, + "open": candle.open, + "high": candle.high, + "low": candle.low, + "close": candle.close, + "volume": candle.volume, + } + for candle in candles + ] + frame = pd.DataFrame.from_records(rows) + if frame.empty: + return pd.DataFrame( + columns=["symbol", "open", "high", "low", "close", "volume"], + index=pd.Index([], name="date"), + ) + + frame["date"] = pd.to_datetime(frame["date"]).dt.date + return frame.set_index("date") + + +def read_existing_candles(path: Path) -> pd.DataFrame: + """Read an existing candle Parquet file as a date-indexed dataframe.""" + + if not path.exists(): + return pd.DataFrame( + columns=["symbol", "open", "high", "low", "close", "volume"], + index=pd.Index([], name="date"), + ) + + frame = pd.read_parquet(path) + if "date" in frame.columns: + frame["date"] = pd.to_datetime(frame["date"]).dt.date + frame = frame.set_index("date") + + frame.index = pd.to_datetime(frame.index).date + frame.index.name = "date" + return frame + + +def oldest_stored_date_or_today(path: Path) -> date: + """Return the oldest stored candle date, or today if no data exists yet.""" + + existing = read_existing_candles(path) + if existing.empty: + return current_market_date() + return min(existing.index) + + +def write_candles(path: Path, symbol: str, candles: list[DailyCandle]) -> pd.DataFrame: + """Append candles to a ticker Parquet file, keeping one row per date.""" + + existing = read_existing_candles(path) + fetched = candles_to_frame(symbol, candles) + combined = pd.concat([existing, fetched]) + if not combined.empty: + combined = combined[~combined.index.duplicated(keep="last")] + combined = combined.sort_index() + + path.parent.mkdir(parents=True, exist_ok=True) + combined.to_parquet(path, index=True) + return combined + + +def print_candles(symbol: str, candles: list[DailyCandle]) -> None: """Print candles in a compact table.""" - print("date,ticker,open,high,low,close,volume") + print("date,symbol,open,high,low,close,volume") for candle in candles: print( f"{candle.trading_day.isoformat()}," - f"{TICKER}," + f"{symbol}," f"{candle.open:.2f}," f"{candle.high:.2f}," f"{candle.low:.2f}," @@ -108,12 +230,63 @@ def print_candles(candles: list[DailyCandle]) -> None: ) -def main() -> None: - """Run the hard-coded fetcher skeleton.""" +def parse_args() -> argparse.Namespace: + """Parse command line arguments.""" - print(f"Fetching {TICKER} daily candles from {START_DATE} to {END_DATE}") - candles = fetch_hard_coded_daily_candles() - print_candles(candles) + parser = argparse.ArgumentParser(description="Fetch daily IBKR candles.") + parser.add_argument("symbol", help="Ticker symbol to fetch, such as SPY.") + end_date_group = parser.add_mutually_exclusive_group() + end_date_group.add_argument( + "--end-date", + type=parse_end_date, + help="Request end date in YYYYMMDD format. Defaults to yesterday.", + ) + end_date_group.add_argument( + "--end-date-from-parquet", + action="store_true", + help=( + "Use the oldest date from the symbol Parquet file as the request " + "end date. Defaults to today if the file is missing or empty." + ), + ) + parser.add_argument( + "--duration", + type=parse_duration, + default=DEFAULT_DURATION, + help=( + "IBKR duration string, such as '1 W' or '1 M'. " + f"Defaults to {DEFAULT_DURATION}." + ), + ) + parser.add_argument( + "--output-dir", + type=Path, + default=DEFAULT_OUTPUT_DIR, + help=f"Directory for Parquet files. Defaults to {DEFAULT_OUTPUT_DIR}.", + ) + return parser.parse_args() + + +def main() -> None: + """Run the daily candle fetcher.""" + + args = parse_args() + symbol = args.symbol.upper() + output_path = args.output_dir / f"{symbol}.parquet" + end_date = ( + oldest_stored_date_or_today(output_path) + if args.end_date_from_parquet + else args.end_date or default_end_date() + ) + + print( + f"Fetching {symbol} daily candles ending {end_date:%Y-%m-%d} " + f"for duration {args.duration}" + ) + candles = fetch_daily_candles(symbol, end_date, args.duration) + print_candles(symbol, candles) + stored = write_candles(output_path, symbol, candles) + print(f"Wrote {len(stored)} total daily rows to {output_path}") if __name__ == "__main__":