Data fetcher implementation

This commit is contained in:
2026-07-26 14:18:07 +03:00
parent 8d3e663e71
commit e2eb5dd574
4 changed files with 273 additions and 58 deletions
+2
View File
@@ -7,3 +7,5 @@ __pycache__/
dist/
build/
*.egg-info/
data/ibkr/
+29 -11
View File
@@ -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.
+34 -12
View File
@@ -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.
+208 -35
View File
@@ -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__":