From ddcf02774ce7254ed0c6d552390754744ecc45b5 Mon Sep 17 00:00:00 2001 From: Sakura <32687351+zifanzhou1024@users.noreply.github.com> Date: Mon, 18 May 2026 03:35:17 -0700 Subject: [PATCH] Add Moomoo ML research recorder --- .gitignore | 1 + README.md | 186 ++++- docs/ml-moomoo-research-recorder.md | 334 +++++++++ ops/run_moomoo_research_market.sh | 45 ++ package.json | 2 + .../moomoo_research_recorder.py | 684 ++++++++++++++++++ .../tests/test_moomoo_research_recorder.py | 259 +++++++ tests/package-scripts.test.mjs | 14 + 8 files changed, 1519 insertions(+), 6 deletions(-) create mode 100644 docs/ml-moomoo-research-recorder.md create mode 100755 ops/run_moomoo_research_market.sh create mode 100644 services/collector/gammascope_collector/moomoo_research_recorder.py create mode 100644 services/collector/tests/test_moomoo_research_recorder.py diff --git a/.gitignore b/.gitignore index 4c43225..e70dc63 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ # Local/editor .worktrees/ .gammascope/ +local-ml-data/ .idea/ .vscode/ .superpowers/ diff --git a/README.md b/README.md index c1ce49b..0791af5 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,32 @@ # GammaScope +GammaScope is a local-first SPX/0DTE analytics workspace with a FastAPI backend, +a Next.js dashboard, collector adapters, replay storage, and a new local ML +research recorder. The current live-source direction is Moomoo OpenD. IBKR +commands remain in the repo as local smoke/probe tools, but new collection work +should start from the Moomoo sections below. + +Current high-level surfaces: + +- Web dashboard: `http://localhost:3000` +- API: `http://127.0.0.1:8000` +- Public live snapshot smoke endpoint: `GET /api/spx/0dte/snapshot/latest` +- Moomoo dashboard collector: `pnpm collector:moomoo-snapshot` +- Moomoo ML research recorder: `pnpm collector:moomoo-research-record` +- Automatic market-hours recorder wrapper: + `ops/run_moomoo_research_market.sh` +- Local ML data root: + `/Users/sakura/local-ml-data/gamma-ml-research` + +Detailed Moomoo ML recorder documentation is in +[docs/ml-moomoo-research-recorder.md](docs/ml-moomoo-research-recorder.md). +Remote AMH/Nginx deployment notes are in +[docs/amh-nginx-server-setup.md](docs/amh-nginx-server-setup.md). + ## Local Development -GammaScope is being built in slices. The first slice establishes the local monorepo, shared contracts, seeded replay data, and smoke-testable API/web surfaces. +GammaScope uses a local pnpm monorepo plus a Python virtualenv for API and +collector code. Project-specific setup notes: Deployment notes for the current Moomoo-backed dashboard and heatmap stack are in [docs/deployment.md](docs/deployment.md). For the AMH/Nginx remote server layout where your computer publishes Moomoo data to a server-hosted backend and frontend, use [docs/amh-nginx-server-setup.md](docs/amh-nginx-server-setup.md). @@ -16,7 +40,7 @@ Run: .venv/bin/python -m pip install -e "apps/api[dev]" pnpm test -### First Slice Verification +### Verification pnpm install pnpm contracts:validate @@ -34,15 +58,86 @@ Run local services: pnpm dev:web .venv/bin/python -m uvicorn gammascope_api.main:app --reload --app-dir apps/api -Open the local dashboard at `http://localhost:3000`. The current dashboard is seeded replay data, shaped to match the live SPX 0DTE analytics contract. +Open the local dashboard at `http://localhost:3000`. The dashboard reads the +stable SPX 0DTE analytics contract. With live collector state available it shows +live mode; otherwise it falls back to seeded replay data. + +## Data And Recorder Map + +The repo now has two separate Moomoo collection paths: + +| Purpose | Command | Storage/Output | +| --- | --- | --- | +| Dashboard compatibility collector | `pnpm collector:moomoo-snapshot -- --publish` | Publishes SPX-shaped collector events to the FastAPI ingestion path. | +| Local ML raw recorder | `pnpm collector:moomoo-research-record` | Writes append-only JSONL under `/Users/sakura/local-ml-data/gamma-ml-research/moomoo/`. | + +The ML recorder is raw-first. It records 0DTE ATM-window option snapshots for: + +```text +SPX, SPY, QQQ, IWM, RUT, NDX +``` + +Rows use the shared future training labels: + +```text +time_utc +time_bucket_utc +ticker +``` + +Compatibility aliases are also written: + +```text +captured_at_utc == time_utc +timeline_bucket_utc == time_bucket_utc +symbol == ticker +``` + +Canonical ticker names are provider-independent. Moomoo codes such as +`US..SPX`, option families such as `SPXW`, and contract codes stay in +source-specific fields such as `owner_code`, `option_code`, or `raw`; they do +not replace top-level `ticker`. + +Local ML file layout: + +```text +/Users/sakura/local-ml-data/gamma-ml-research/ + moomoo/ + options/date=YYYY-MM-DD/ticker=SPX/records.jsonl + underlyings/date=YYYY-MM-DD/ticker=SPX/records.jsonl + batches/date=YYYY-MM-DD/ticker=SPX/records.jsonl + captures/date=YYYY-MM-DD/captures.jsonl + gexbot/ + responses/date=YYYY-MM-DD/ticker=SPX/endpoint=/records.jsonl + captures/date=YYYY-MM-DD/captures.jsonl + logs/ +``` + +The `gexbot/` folders are reserved for the companion Dealer Flow Lab recorder. +The implementation handoff for that side lives in: + +```text +/Users/sakura/WebstormProjects/dealer-flow-lab/docs/gexbot-ml-local-recorder-handoff.md +``` + +Future preprocessing should join Moomoo and GEXBot with: + +```text +market_date + ticker + time_bucket_utc +``` + +Use backward/as-of joins only. Do not create model labels or train/test splits in +the raw collectors. ### Mock Local Collector -Before connecting to IBKR, the collector slice can emit a deterministic SPX 0DTE event cycle as newline-delimited JSON: +For deterministic local smoke tests, the mock collector can emit an SPX 0DTE +event cycle as newline-delimited JSON: pnpm collector:mock -- --spot 5200.25 --expiry 2026-04-23 --strikes 5190,5200,5210 -The mock output uses the same normalized collector event contract planned for the live IBKR adapter. +The mock output uses the same normalized collector event contract consumed by +the local ingestion path. ### Local Collector Ingestion @@ -188,7 +283,10 @@ Then open `http://localhost:3000`, use the replay controls, and pick the capture ### Local Moomoo 0DTE Snapshot -Moomoo is the default direction for new live-source work. The first Moomoo collector uses local OpenD and keeps the current SPX dashboard contract by publishing only SPX rows into the existing collector event path. +Moomoo is the default direction for new live-source work. The dashboard collector +uses local OpenD and keeps the current SPX dashboard contract by publishing only +SPX rows into the existing collector event path. The separate ML research +recorder below captures the wider configured universe to local JSONL files. Install the Moomoo package in the project virtualenv: @@ -210,6 +308,82 @@ Publish SPX compatibility events into the local FastAPI ingestion path. By defau The collector fetches the configured universe: SPX, SPY, QQQ, IWM, RUT, and NDX. It polls `get_market_snapshot()` every 2 seconds during active market/pre-open windows, reduces to once per minute from 5:00 PM to 8:30 AM New York time, refreshes the SPX spot proxy every loop, infers the SPX spot from same-strike call/put mids, and chunks requests to at most 400 option codes. The default expiry is chosen in New York time: today's 0DTE until 4:05 PM, then the next weekday session so expired 0DTE chains are not reused overnight. Pass `--expiry YYYY-MM-DD` only when you intentionally want to pin a smoke test to a specific expiry. It uses `get_option_chain()` at startup and again if the automatic expiry changes while the collector is running. +### Local Moomoo ML Research Recorder + +Use the research recorder when the goal is future model training rather than +dashboard display. It preserves the complete Moomoo snapshot row in `raw` and +adds stable labels for later timeline alignment. + +One-loop smoke capture: + + pnpm collector:moomoo-research-record -- --max-loops 1 + +Continuous 10-second capture: + + pnpm collector:moomoo-research-record + +Market-hours capture that waits for 9:30 AM Eastern, exits after 4:00 PM +Eastern, and repeats weekdays while the process stays alive: + + pnpm collector:moomoo-research-market + +RUT and NDX currently need manual spot values before option rows can be selected: + + pnpm collector:moomoo-research-record -- --spot RUT=2150 --spot NDX=18400 + +The default output root is: + + /Users/sakura/local-ml-data/gamma-ml-research + +The recorder writes four Moomoo families: + + moomoo/options/date=YYYY-MM-DD/ticker=SPX/records.jsonl + moomoo/underlyings/date=YYYY-MM-DD/ticker=SPX/records.jsonl + moomoo/batches/date=YYYY-MM-DD/ticker=SPX/records.jsonl + moomoo/captures/date=YYYY-MM-DD/captures.jsonl + +`options` is one row per option contract per capture. `batches` is one row per +ticker per capture, with all returned option contracts nested under +`contracts[]`. Use `raw` or `contracts[].raw` as the source of truth for future +feature engineering. + +### Automatic Market-Hours Recording + +Automatic Moomoo ML collection is configured through a macOS LaunchAgent: + + /Users/sakura/Library/LaunchAgents/com.sakura.gammascope.moomoo-research-recorder.plist + +It runs: + + /Users/sakura/WebstormProjects/gamma-scope/ops/run_moomoo_research_market.sh + +The wrapper starts the market-hours recorder, waits for the regular U.S. session +when needed, repeats across weekdays, and restarts after 5 minutes if the +recorder exits. Logs are written to: + + /Users/sakura/local-ml-data/gamma-ml-research/logs/moomoo-research-recorder.out.log + /Users/sakura/local-ml-data/gamma-ml-research/logs/moomoo-research-recorder.err.log + +Check service status: + + launchctl print gui/$(id -u)/com.sakura.gammascope.moomoo-research-recorder + +Restart after changing recorder arguments: + + launchctl kickstart -k gui/$(id -u)/com.sakura.gammascope.moomoo-research-recorder + +Stop automatic collection: + + launchctl bootout gui/$(id -u)/com.sakura.gammascope.moomoo-research-recorder + +Optional extra arguments live in: + + /Users/sakura/local-ml-data/gamma-ml-research/moomoo-recorder.args + +Use that file for RUT/NDX manual spots or a 30-second fallback cadence. The +LaunchAgent cannot wake a sleeping Mac, and Moomoo OpenD still needs to be +available and logged in. + ### SPX 0DTE Exposure Heatmap The latest-ladder heatmap page is available at `http://localhost:3000/heatmap` when the web app is running. diff --git a/docs/ml-moomoo-research-recorder.md b/docs/ml-moomoo-research-recorder.md new file mode 100644 index 0000000..a1168e2 --- /dev/null +++ b/docs/ml-moomoo-research-recorder.md @@ -0,0 +1,334 @@ +# Moomoo ML Research Recorder + +Date: 2026-05-18 +Repository: gamma-scope + +## Purpose + +GammaScope has a local-only Moomoo recorder for machine-learning research. It records raw 0DTE option snapshots for the configured ATM-window universe without changing the live dashboard contract. + +The default output folder is: + +```text +~/local-ml-data/gamma-ml-research +``` + +This folder is intentionally outside the repo so large local data is not committed. If you override the output path into the repo, `local-ml-data/` is ignored by git. + +## Source Coverage + +The recorder uses the same configured Moomoo universe as the collector: + +```text +SPX, SPY, QQQ, IWM, RUT, NDX +``` + +It still selects only the active 0DTE expiry and ATM-centered strike windows. RUT and NDX require manual spot values unless a later source improves index spot resolution: + +```bash +pnpm collector:moomoo-research-record -- --spot RUT=2150 --spot NDX=18400 +``` + +## Ticker Naming Contract + +Use `ticker` as the canonical training identity across every local ML file. It must be uppercase, provider-independent, and must identify the underlying research ticker, not an option contract, provider code, option family, or spot proxy. + +For the current universe, only these top-level ticker names should appear: + +```text +SPX, SPY, QQQ, RUT, IWM, NDX +``` + +Moomoo-specific names stay in source-specific fields: + +| Canonical `ticker` | Moomoo owner code | Moomoo option family filter | Notes | +| --- | --- | --- | --- | +| `SPX` | `US..SPX` | `SPXW` | SPX rows remain `ticker=SPX` even when spot is resolved from the SPY proxy. | +| `SPY` | `US.SPY` | none | ETF ticker maps directly. | +| `QQQ` | `US.QQQ` | none | ETF ticker maps directly. | +| `IWM` | `US.IWM` | none | ETF ticker maps directly. | +| `RUT` | `US..RUT` | `RUTW` | Requires manual spot today. | +| `NDX` | `US..NDX` | `NDXP` | Requires manual spot today. | + +Rules: + +- Top-level `ticker` and `symbol` must match exactly. `symbol` exists only as a compatibility alias. +- Do not write Moomoo region codes such as `US.SPY` or `US..SPX` into `ticker`. +- Do not write option family names such as `SPXW`, `RUTW`, or `NDXP` into `ticker`. +- Do not write option contract codes into `ticker`; use `option_code` and `raw.code` for provider contract identity. +- Do not change `ticker` when a proxy is used for spot. For example, SPX proxy spot can come from `US.SPY`, but the row remains `ticker=SPX`. + +## Run Commands + +One-loop smoke capture: + +```bash +pnpm collector:moomoo-research-record -- --max-loops 1 +``` + +Continuous 10-second capture: + +```bash +pnpm collector:moomoo-research-record +``` + +One regular market session only. If started before 9:30 AM Eastern, it waits for the next weekday open; it exits after 4:00 PM Eastern: + +```bash +pnpm collector:moomoo-research-record -- --market-hours +``` + +Repeat every weekday regular session while the process stays running: + +```bash +pnpm collector:moomoo-research-market +``` + +Automatic macOS LaunchAgent: + +```text +~/Library/LaunchAgents/com.sakura.gammascope.moomoo-research-recorder.plist +``` + +This LaunchAgent starts: + +```text +/Users/sakura/WebstormProjects/gamma-scope/ops/run_moomoo_research_market.sh +``` + +The wrapper runs the market-hours recorder continuously, waits for regular market open, repeats across weekdays, and restarts with a 5-minute backoff if the recorder exits. It writes logs to: + +```text +/Users/sakura/local-ml-data/gamma-ml-research/logs/moomoo-research-recorder.out.log +/Users/sakura/local-ml-data/gamma-ml-research/logs/moomoo-research-recorder.err.log +``` + +Check status: + +```bash +launchctl print gui/$(id -u)/com.sakura.gammascope.moomoo-research-recorder +``` + +Restart after editing args: + +```bash +launchctl kickstart -k gui/$(id -u)/com.sakura.gammascope.moomoo-research-recorder +``` + +Stop automatic collection: + +```bash +launchctl bootout gui/$(id -u)/com.sakura.gammascope.moomoo-research-recorder +``` + +Extra arguments can be placed in: + +```text +/Users/sakura/local-ml-data/gamma-ml-research/moomoo-recorder.args +``` + +Use that file for RUT/NDX manual spots or a 30-second fallback cadence. + +Continuous capture with manual RUT/NDX spots: + +```bash +pnpm collector:moomoo-research-record -- --spot RUT=2150 --spot NDX=18400 +``` + +Custom output folder: + +```bash +pnpm collector:moomoo-research-record -- --output-dir ~/local-ml-data/gamma-ml-research +``` + +The default cadence is 10 seconds. Use 30 seconds only when collection stability or later feature processing requires it: + +```bash +pnpm collector:moomoo-research-record -- --interval-seconds 30 --timeline-seconds 30 +``` + +Market-hours mode uses the normal U.S. regular session: + +```text +09:30 AM to 04:00 PM America/New_York +``` + +It skips weekends. It does not yet embed a full exchange-holiday calendar; on an exchange holiday it may wake during the regular window and record empty or degraded captures unless the process is stopped. + +## File Layout + +Option rows: + +```text +~/local-ml-data/gamma-ml-research/moomoo/options/date=YYYY-MM-DD/ticker=SPX/records.jsonl +``` + +Underlying/proxy rows: + +```text +~/local-ml-data/gamma-ml-research/moomoo/underlyings/date=YYYY-MM-DD/ticker=SPX/records.jsonl +``` + +Batch rows, one line per capture and ticker with all returned option contracts nested under `contracts`: + +```text +~/local-ml-data/gamma-ml-research/moomoo/batches/date=YYYY-MM-DD/ticker=SPX/records.jsonl +``` + +Capture summaries: + +```text +~/local-ml-data/gamma-ml-research/moomoo/captures/date=YYYY-MM-DD/captures.jsonl +``` + +## File Readability Contract + +The files are JSONL: one complete JSON object per line. This is deliberate: + +- Easy to append safely if the recorder stops and restarts. +- Easy to inspect with `head`, `jq`, Python, pandas, or PyTorch data loaders. +- Easy to convert later into Parquet after the raw capture is trusted. + +Read one day of SPX options in Python: + +```python +import json +from pathlib import Path + +path = Path("/Users/sakura/local-ml-data/gamma-ml-research/moomoo/options/date=2026-05-19/ticker=SPX/records.jsonl") +rows = [json.loads(line) for line in path.read_text().splitlines()] +``` + +Read with pandas: + +```python +import pandas as pd + +df = pd.read_json( + "/Users/sakura/local-ml-data/gamma-ml-research/moomoo/options/date=2026-05-19/ticker=SPX/records.jsonl", + lines=True, +) +``` + +## Record Labeling + +The recorder labels rows for future training, but it does not create model target labels yet. + +Top-level fields are intentionally stable: + +| Field | Meaning | +| --- | --- | +| `schema_version` | Recorder schema version. | +| `source` | Always `moomoo`. | +| `record_type` | `option_snapshot`, `option_snapshot_batch`, `underlying_snapshot`, or `capture_summary`. | +| `capture_id` | Unique capture-loop id. Rows with the same id came from the same recorder cycle. | +| `time_utc` | Canonical local recorder receipt time in UTC. Use this for as-of joins. | +| `captured_at_utc` | Backward-compatible alias for `time_utc`. | +| `time_bucket_utc` | Canonical capture time floored to the configured 10-second or 30-second grid. | +| `timeline_bucket_utc` | Backward-compatible alias for `time_bucket_utc`. | +| `timeline_seconds` | Grid size used to calculate `time_bucket_utc`. Defaults to `10`. | +| `market_date` | New York market date. | +| `ticker` | Canonical training ticker such as `SPX`, `SPY`, `QQQ`, `IWM`, `RUT`, or `NDX`. | +| `symbol` | Backward-compatible alias for `ticker`. | +| `expiry` | Target 0DTE option expiry. | +| `option_code` | Moomoo option contract code. Option rows only. | +| `strike` | Contract strike as a number. Option rows only. | +| `right` | `call` or `put`. Option rows only. | +| `provider_update_time` | Moomoo's update time when supplied. Do not use it as the primary timeline key. | +| `raw` | Full provider payload. This is the source of truth. | +| `normalized` | Convenience projection used by GammaScope today. Do not treat it as the full dataset. | + +The primary training-time key should be: + +```text +source + market_date + ticker + expiry + time_utc + option_code +``` + +For alignment across Moomoo and GEXBot, use: + +```text +market_date + ticker + time_bucket_utc +``` + +Do not interpret `time_bucket_utc` as proof that the provider updated exactly at that second. It is a capture-grid label for joining and gap detection. + +Batch rows use the same top-level labels. Each batch row has: + +| Field | Meaning | +| --- | --- | +| `record_type` | Always `option_snapshot_batch`. | +| `ticker` | Canonical ticker for this capture batch. | +| `contract_count` | Number of option contracts returned by Moomoo for that ticker in this capture. | +| `contracts` | Nested list of returned contracts. Each item includes `option_code`, `strike`, `right`, `provider_update_time`, `raw`, and `normalized`. | + +## Raw-First Rule + +Do not drop, rename, or reshape Moomoo raw fields at collection time. The current recorder keeps fields such as: + +- `option_net_open_interest` +- `option_premium` +- `option_contract_nominal_value` +- `option_expiry_date_distance` +- `bid_ask_ratio` +- pre-market, after-hours, and overnight fields +- any future unknown provider fields + +Later preprocessing can decide what to keep. The recorder's job is to preserve enough information to re-run feature engineering without re-collecting the day. + +Each row has UTC timing fields for later synchronization: + +- `time_utc` +- `time_bucket_utc` +- `captured_at_utc` +- `timeline_bucket_utc` +- `market_date` +- `expiry` +- `ticker` +- `symbol` +- `option_code` +- `strike` +- `right` +- `provider_update_time` + +Each option row also stores: + +- `raw`: the full Moomoo `get_market_snapshot()` row. +- `normalized`: GammaScope's current normalized convenience fields. + +Each batch row stores: + +- `contracts[].raw`: the full Moomoo `get_market_snapshot()` row for each returned contract in that capture. +- `contracts[].normalized`: the same convenience projection as the flat option rows. + +For ML, treat `raw` as the source of truth. The normalized block is only for easy inspection and compatibility. + +## Timeline Alignment + +The intended training dataset should be built from raw files using a fixed exchange-time grid: + +```text +09:30:00 ET +09:30:10 ET +09:30:20 ET +... +16:00:00 ET +``` + +For each timeline row at time `t`, use only rows with `time_utc <= t`. + +Recommended join tolerances: + +- Moomoo: 15 seconds on a 10-second grid. +- GEXBot: 2 to 5 minutes, depending on observed update speed. + +Do not use nearest joins that can pull future data. Use backward/as-of joins only. + +## Modeling Note + +Train in stages: + +1. Moomoo-only baseline model. +2. GEXBot-only baseline model. +3. Fused model using the aligned timeline. + +This prevents a weak or stale source from hiding inside a blended model. Keep the fused model only if walk-forward tests show it improves out-of-sample performance. diff --git a/ops/run_moomoo_research_market.sh b/ops/run_moomoo_research_market.sh new file mode 100755 index 0000000..55e7f8a --- /dev/null +++ b/ops/run_moomoo_research_market.sh @@ -0,0 +1,45 @@ +#!/bin/zsh +set -u + +export PATH="/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin" + +REPO_DIR="/Users/sakura/WebstormProjects/gamma-scope" +DATA_DIR="/Users/sakura/local-ml-data/gamma-ml-research" +LOG_DIR="${DATA_DIR}/logs" +ARGS_FILE="${DATA_DIR}/moomoo-recorder.args" + +mkdir -p "${LOG_DIR}" + +cd "${REPO_DIR}" || { + echo "$(date -u '+%Y-%m-%dT%H:%M:%SZ') unable to cd into ${REPO_DIR}" >&2 + exit 1 +} + +load_extra_args() { + local line + EXTRA_ARGS=() + if [[ ! -f "${ARGS_FILE}" ]]; then + return + fi + while IFS= read -r line || [[ -n "${line}" ]]; do + [[ -z "${line//[[:space:]]/}" ]] && continue + [[ "${line}" == \#* ]] && continue + EXTRA_ARGS+=(${(z)line}) + done < "${ARGS_FILE}" +} + +while true; do + load_extra_args + echo "$(date -u '+%Y-%m-%dT%H:%M:%SZ') starting moomoo research market recorder" + echo "$(date -u '+%Y-%m-%dT%H:%M:%SZ') extra args: ${EXTRA_ARGS[*]:-(none)}" + + PYTHONPATH="services/collector:apps/api" \ + .venv/bin/python -m gammascope_collector.moomoo_research_recorder \ + --market-hours \ + --repeat-daily \ + "${EXTRA_ARGS[@]}" + + status=$? + echo "$(date -u '+%Y-%m-%dT%H:%M:%SZ') recorder exited with status ${status}; restarting after 300 seconds" >&2 + sleep 300 +done diff --git a/package.json b/package.json index eaff73d..0bddd40 100644 --- a/package.json +++ b/package.json @@ -18,6 +18,8 @@ "collector:ibkr-contracts": "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.ibkr_contracts", "collector:ibkr-delayed-snapshot": "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.ibkr_delayed_snapshot", "collector:moomoo-snapshot": "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.moomoo_snapshot", + "collector:moomoo-research-record": "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.moomoo_research_recorder", + "collector:moomoo-research-market": "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.moomoo_research_recorder --market-hours --repeat-daily", "collector:publish-mock": "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.publisher", "test:collector": "PYTHONPATH=services/collector:apps/api .venv/bin/pytest services/collector/tests -q", "test": "pnpm test:scripts && pnpm test:contracts && pnpm test:api && pnpm typecheck:web && pnpm test:web" diff --git a/services/collector/gammascope_collector/moomoo_research_recorder.py b/services/collector/gammascope_collector/moomoo_research_recorder.py new file mode 100644 index 0000000..0b82a49 --- /dev/null +++ b/services/collector/gammascope_collector/moomoo_research_recorder.py @@ -0,0 +1,684 @@ +from __future__ import annotations + +import argparse +import json +import math +import sys +import time +from collections.abc import Callable, Sequence +from dataclasses import asdict, dataclass +from datetime import UTC, date, datetime, time as wall_time, timedelta +from pathlib import Path +from typing import Any + +from gammascope_collector.moomoo_config import ( + DEFAULT_MOOMOO_HOST, + DEFAULT_MOOMOO_PORT, + MoomooCollectorConfig, + MoomooSymbolConfig, + chunked, + parse_manual_spots, + selected_symbols, +) +from gammascope_collector.moomoo_snapshot import ( + MARKET_TIMEZONE, + RET_OK, + MoomooContract, + MoomooQuoteClient, + MoomooSymbolDiscoveryResult, + _normalize_record, + _records, + _target_expiry, + discover_symbol_contracts, + normalize_snapshot_record, + resolve_moomoo_target_expiry, +) + +DEFAULT_RESEARCH_OUTPUT_DIR = Path.home() / "local-ml-data" / "gamma-ml-research" +DEFAULT_RESEARCH_INTERVAL_SECONDS = 10.0 +DEFAULT_TIMELINE_SECONDS = 10 +DEFAULT_MARKET_OPEN = wall_time(9, 30) +DEFAULT_MARKET_CLOSE = wall_time(16, 0) +SCHEMA_VERSION = "1.0.0" + +ClientFactory = Callable[[str, int], MoomooQuoteClient] +ExpiryProvider = Callable[[], date] +IntervalSecondsProvider = Callable[[], float] + + +@dataclass(frozen=True) +class ResearchCaptureSummary: + capture_id: str + time_utc: str + captured_at_utc: str + time_bucket_utc: str + market_date: str + expiry: str + output_dir: str + option_rows_written: int + batch_rows_written: int + underlying_rows_written: int + selected_contracts: dict[str, int] + option_files: list[str] + batch_files: list[str] + underlying_files: list[str] + warnings: list[str] + + def as_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class MarketSessionWindow: + market_date: str + opens_at_utc: str + closes_at_utc: str + + +def capture_research_snapshot_once( + client: MoomooQuoteClient, + config: MoomooCollectorConfig, + *, + output_dir: Path, + expiry: date, + timeline_seconds: int = DEFAULT_TIMELINE_SECONDS, +) -> ResearchCaptureSummary: + if timeline_seconds <= 0: + raise ValueError("timeline_seconds must be greater than zero") + + output_dir = output_dir.expanduser() + captured_at = datetime.now(UTC) + capture_id = _capture_id(captured_at) + timeline_bucket = _floor_datetime(captured_at, timeline_seconds) + time_utc = _format_datetime(captured_at) + time_bucket_utc = _format_datetime(timeline_bucket) + market_date = captured_at.astimezone(MARKET_TIMEZONE).date().isoformat() + + symbols = selected_symbols(config) + subscription_code, subscription = client.query_subscription(is_all_conn=True) + warnings: list[str] = [] + if subscription_code != RET_OK: + warnings.append(f"subscription query failed with code {subscription_code}") + + discoveries = [discover_symbol_contracts(client, symbol, expiry=expiry) for symbol in symbols] + warnings.extend(warning for discovery in discoveries for warning in discovery.warnings) + discovery_by_symbol = {discovery.symbol: discovery for discovery in discoveries} + + underlying_summary = _record_underlying_snapshots( + client=client, + symbols=symbols, + discoveries=discovery_by_symbol, + output_dir=output_dir, + capture_id=capture_id, + captured_at=captured_at, + timeline_bucket=timeline_bucket, + timeline_seconds=timeline_seconds, + market_date=market_date, + expiry=expiry, + ) + warnings.extend(underlying_summary.warnings) + + contract_by_code: dict[str, MoomooContract] = {} + for discovery in discoveries: + for contract in discovery.contracts: + contract_by_code[contract.option_code] = contract + + option_rows_written = 0 + option_files: set[Path] = set() + batch_contracts_by_ticker: dict[str, list[dict[str, object]]] = {} + batch_spot_by_ticker: dict[str, float | None] = {} + returned_codes: set[str] = set() + for code_chunk in chunked(sorted(contract_by_code), 400): + return_code, snapshot_data = client.get_market_snapshot(code_chunk) + if return_code != RET_OK: + warnings.append(f"option snapshot request failed with code {return_code}") + continue + for raw_record in _records(snapshot_data): + normalized_record = _normalize_record(raw_record) + option_code = str(normalized_record.get("code") or "") + contract = contract_by_code.get(option_code) + if contract is None: + continue + returned_codes.add(option_code) + discovery = discovery_by_symbol.get(contract.symbol) + record = _option_output_record( + raw_record=normalized_record, + contract=contract, + discovery=discovery, + capture_id=capture_id, + captured_at=captured_at, + timeline_bucket=timeline_bucket, + timeline_seconds=timeline_seconds, + market_date=market_date, + expiry=expiry, + ) + path = _records_path(output_dir, "moomoo", "options", market_date, contract.symbol) + _append_jsonl(path, record) + option_files.add(path) + option_rows_written += 1 + ticker = _canonical_ticker(contract.symbol) + batch_contracts_by_ticker.setdefault(ticker, []).append(_batch_contract_record(record)) + batch_spot_by_ticker[ticker] = discovery.spot if discovery is not None else None + + missing_count = len(contract_by_code) - len(returned_codes) + if missing_count > 0: + warnings.append(f"option snapshot missing {missing_count} selected contracts") + + batch_files: set[Path] = set() + batch_rows_written = 0 + for ticker, contracts in sorted(batch_contracts_by_ticker.items()): + batch_record = { + "schema_version": SCHEMA_VERSION, + "source": "moomoo", + "record_type": "option_snapshot_batch", + "capture_id": capture_id, + "time_utc": time_utc, + "captured_at_utc": time_utc, + "time_bucket_utc": time_bucket_utc, + "timeline_bucket_utc": time_bucket_utc, + "timeline_seconds": timeline_seconds, + "market_date": market_date, + "ticker": ticker, + "symbol": ticker, + "expiry": expiry.isoformat(), + "resolved_spot": batch_spot_by_ticker.get(ticker), + "contract_count": len(contracts), + "contracts": contracts, + } + path = _batch_records_path(output_dir, "moomoo", market_date, ticker) + _append_jsonl(path, batch_record) + batch_files.add(path) + batch_rows_written += 1 + + _append_jsonl( + _captures_path(output_dir, market_date), + { + "schema_version": SCHEMA_VERSION, + "source": "moomoo", + "record_type": "capture_summary", + "capture_id": capture_id, + "time_utc": time_utc, + "captured_at_utc": time_utc, + "time_bucket_utc": time_bucket_utc, + "timeline_bucket_utc": time_bucket_utc, + "timeline_seconds": timeline_seconds, + "market_date": market_date, + "expiry": expiry.isoformat(), + "subscription": _jsonable(subscription), + "option_rows_written": option_rows_written, + "batch_rows_written": batch_rows_written, + "underlying_rows_written": underlying_summary.rows_written, + "selected_contracts": {discovery.symbol: len(discovery.contracts) for discovery in discoveries}, + "tickers": [_canonical_ticker(discovery.symbol) for discovery in discoveries], + "warnings": warnings, + }, + ) + + return ResearchCaptureSummary( + capture_id=capture_id, + time_utc=time_utc, + captured_at_utc=time_utc, + time_bucket_utc=time_bucket_utc, + market_date=market_date, + expiry=expiry.isoformat(), + output_dir=str(output_dir), + option_rows_written=option_rows_written, + batch_rows_written=batch_rows_written, + underlying_rows_written=underlying_summary.rows_written, + selected_contracts={discovery.symbol: len(discovery.contracts) for discovery in discoveries}, + option_files=[str(path) for path in sorted(option_files)], + batch_files=[str(path) for path in sorted(batch_files)], + underlying_files=[str(path) for path in sorted(underlying_summary.files)], + warnings=warnings, + ) + + +def run_research_recorder_loop( + client: MoomooQuoteClient, + config: MoomooCollectorConfig, + *, + output_dir: Path, + expiry: date | None, + expiry_provider: ExpiryProvider | None = None, + interval_seconds_provider: IntervalSecondsProvider | None = None, + timeline_seconds: int = DEFAULT_TIMELINE_SECONDS, + max_loops: int | None = None, +) -> ResearchCaptureSummary: + result: ResearchCaptureSummary | None = None + loops = 0 + while max_loops is None or loops < max_loops: + target_expiry = _target_expiry(expiry, expiry_provider) + started_at = time.perf_counter() + result = capture_research_snapshot_once( + client, + config, + output_dir=output_dir, + expiry=target_expiry, + timeline_seconds=timeline_seconds, + ) + print(json.dumps(result.as_dict(), separators=(",", ":"), sort_keys=True), flush=True) + elapsed = time.perf_counter() - started_at + loops += 1 + if max_loops is None or loops < max_loops: + interval_seconds = interval_seconds_provider() if interval_seconds_provider else config.refresh_interval_seconds + time.sleep(max(0, interval_seconds - elapsed)) + + if result is None: + raise RuntimeError("research recorder loop did not run") + return result + + +def run_market_hours_recorder_loop( + client: MoomooQuoteClient, + config: MoomooCollectorConfig, + *, + output_dir: Path, + expiry: date | None, + expiry_provider: ExpiryProvider | None = None, + interval_seconds_provider: IntervalSecondsProvider | None = None, + timeline_seconds: int = DEFAULT_TIMELINE_SECONDS, + max_loops: int | None = None, + repeat_daily: bool = False, + now_provider: Callable[[], datetime] | None = None, + sleep: Callable[[float], None] | None = None, +) -> ResearchCaptureSummary: + if timeline_seconds <= 0: + raise ValueError("timeline_seconds must be greater than zero") + + current_time = now_provider or (lambda: datetime.now(UTC)) + sleeper = sleep or time.sleep + result: ResearchCaptureSummary | None = None + loops = 0 + completed_session_dates: set[str] = set() + + while max_loops is None or loops < max_loops: + now = _aware_utc(current_time()) + session = next_regular_session_window(now) + opens_at = _parse_datetime(session.opens_at_utc) + closes_at = _parse_datetime(session.closes_at_utc) + + if now < opens_at: + _print_market_status("waiting_for_market_open", session, now) + sleeper(min(60.0, max(0.0, (opens_at - now).total_seconds()))) + continue + if now >= closes_at: + completed_session_dates.add(session.market_date) + if not repeat_daily: + break + _print_market_status("waiting_for_next_market_day", next_regular_session_window(now + timedelta(seconds=1)), now) + sleeper(60.0) + continue + + target_expiry = _target_expiry(expiry, expiry_provider) + started_at = time.perf_counter() + result = capture_research_snapshot_once( + client, + config, + output_dir=output_dir, + expiry=target_expiry, + timeline_seconds=timeline_seconds, + ) + print(json.dumps(result.as_dict(), separators=(",", ":"), sort_keys=True), flush=True) + loops += 1 + + now_after_capture = _aware_utc(current_time()) + if now_after_capture >= closes_at: + completed_session_dates.add(session.market_date) + if not repeat_daily or (max_loops is not None and loops >= max_loops): + break + continue + + interval_seconds = interval_seconds_provider() if interval_seconds_provider else config.refresh_interval_seconds + elapsed = time.perf_counter() - started_at + seconds_until_close = max(0.0, (closes_at - now_after_capture).total_seconds()) + sleeper(min(seconds_until_close, max(0.0, interval_seconds - elapsed))) + + if result is None: + raise RuntimeError("market-hours recorder stopped before any capture") + return result + + +def next_regular_session_window(now: datetime | None = None) -> MarketSessionWindow: + market_now = _aware_utc(now or datetime.now(UTC)).astimezone(MARKET_TIMEZONE) + session_date = market_now.date() + if market_now.weekday() >= 5 or market_now.time() >= DEFAULT_MARKET_CLOSE: + session_date = _next_weekday(session_date + timedelta(days=1)) + opens_at = datetime.combine(session_date, DEFAULT_MARKET_OPEN, MARKET_TIMEZONE).astimezone(UTC) + closes_at = datetime.combine(session_date, DEFAULT_MARKET_CLOSE, MARKET_TIMEZONE).astimezone(UTC) + return MarketSessionWindow( + market_date=session_date.isoformat(), + opens_at_utc=_format_datetime(opens_at), + closes_at_utc=_format_datetime(closes_at), + ) + + +def main(argv: Sequence[str] | None = None, *, client_factory: ClientFactory | None = None) -> None: + parser = argparse.ArgumentParser(description="Record raw Moomoo 0DTE option snapshots for local ML research.") + parser.add_argument("--host", default=DEFAULT_MOOMOO_HOST) + parser.add_argument("--port", type=int, default=DEFAULT_MOOMOO_PORT) + parser.add_argument("--output-dir", type=Path, default=DEFAULT_RESEARCH_OUTPUT_DIR) + parser.add_argument( + "--expiry", + type=_parse_date, + default=None, + help="Target option expiry. Defaults to the current/next New York market session.", + ) + parser.add_argument("--spot", action="append", default=[], help="Manual spot override, for example RUT=2150.") + parser.add_argument("--interval-seconds", type=float, default=DEFAULT_RESEARCH_INTERVAL_SECONDS) + parser.add_argument("--timeline-seconds", type=int, default=DEFAULT_TIMELINE_SECONDS) + parser.add_argument("--max-loops", type=int, default=0, help="Number of loops to run; 0 runs continuously.") + parser.add_argument( + "--market-hours", + action="store_true", + help="Wait for the next regular 9:30 AM ET session and stop at 4:00 PM ET.", + ) + parser.add_argument( + "--repeat-daily", + action="store_true", + help="With --market-hours, keep recording each weekday market session.", + ) + raw_args = list(argv if argv is not None else sys.argv[1:]) + if raw_args[:1] == ["--"]: + raw_args = raw_args[1:] + args = parser.parse_args(raw_args) + + client: MoomooQuoteClient | None = None + try: + config = MoomooCollectorConfig( + host=args.host, + port=args.port, + refresh_interval_seconds=args.interval_seconds, + manual_spots=parse_manual_spots(args.spot), + ) + make_client = client_factory or _create_real_client + client = make_client(args.host, args.port) + loop_kwargs = { + "output_dir": args.output_dir, + "expiry": args.expiry, + "expiry_provider": None if args.expiry is not None else resolve_moomoo_target_expiry, + "interval_seconds_provider": lambda: config.refresh_interval_seconds, + "timeline_seconds": args.timeline_seconds, + "max_loops": _normalize_max_loops(args.max_loops), + } + if args.market_hours: + run_market_hours_recorder_loop( + client, + config, + repeat_daily=args.repeat_daily, + **loop_kwargs, + ) + else: + if args.repeat_daily: + raise ValueError("--repeat-daily requires --market-hours") + run_research_recorder_loop( + client, + config, + **loop_kwargs, + ) + except Exception as exc: + print(json.dumps({"status": "error", "message": str(exc)}, sort_keys=True, separators=(",", ":"))) + raise SystemExit(1) from exc + finally: + if client is not None: + client.close() + + +@dataclass(frozen=True) +class _UnderlyingWriteSummary: + rows_written: int + files: set[Path] + warnings: list[str] + + +def _record_underlying_snapshots( + *, + client: MoomooQuoteClient, + symbols: Sequence[MoomooSymbolConfig], + discoveries: dict[str, MoomooSymbolDiscoveryResult], + output_dir: Path, + capture_id: str, + captured_at: datetime, + timeline_bucket: datetime, + timeline_seconds: int, + market_date: str, + expiry: date, +) -> _UnderlyingWriteSummary: + requested_codes: dict[str, list[tuple[str, str]]] = {} + for symbol in symbols: + requested_codes.setdefault(symbol.owner_code, []).append((symbol.symbol, "owner")) + if symbol.spot_proxy_code: + requested_codes.setdefault(symbol.spot_proxy_code, []).append((symbol.symbol, "spot_proxy")) + + records_by_code: dict[str, dict[str, object]] = {} + warnings: list[str] = [] + for provider_code in sorted(requested_codes): + return_code, snapshot_data = client.get_market_snapshot([provider_code]) + if return_code != RET_OK: + warnings.append(f"underlying snapshot request failed for {provider_code} with code {return_code}") + continue + for raw_record in _records(snapshot_data): + normalized_record = _normalize_record(raw_record) + code = str(normalized_record.get("code") or "") + if code: + records_by_code[code] = normalized_record + + rows_written = 0 + files: set[Path] = set() + for provider_code, usages in requested_codes.items(): + raw_record = records_by_code.get(provider_code) + for symbol, usage in usages: + ticker = _canonical_ticker(symbol) + discovery = discoveries.get(symbol) + time_utc = _format_datetime(captured_at) + time_bucket_utc = _format_datetime(timeline_bucket) + record = { + "schema_version": SCHEMA_VERSION, + "source": "moomoo", + "record_type": "underlying_snapshot", + "capture_id": capture_id, + "time_utc": time_utc, + "captured_at_utc": time_utc, + "time_bucket_utc": time_bucket_utc, + "timeline_bucket_utc": time_bucket_utc, + "timeline_seconds": timeline_seconds, + "market_date": market_date, + "ticker": ticker, + "symbol": ticker, + "expiry": expiry.isoformat(), + "provider_code": provider_code, + "usage": usage, + "resolved_spot": discovery.spot if discovery is not None else None, + "raw": _jsonable(raw_record or {}), + } + path = _records_path(output_dir, "moomoo", "underlyings", market_date, symbol) + _append_jsonl(path, record) + rows_written += 1 + files.add(path) + + return _UnderlyingWriteSummary(rows_written=rows_written, files=files, warnings=warnings) + + +def _option_output_record( + *, + raw_record: dict[str, object], + contract: MoomooContract, + discovery: MoomooSymbolDiscoveryResult | None, + capture_id: str, + captured_at: datetime, + timeline_bucket: datetime, + timeline_seconds: int, + market_date: str, + expiry: date, +) -> dict[str, object]: + normalized = normalize_snapshot_record(contract, raw_record) + ticker = _canonical_ticker(contract.symbol) + time_utc = _format_datetime(captured_at) + time_bucket_utc = _format_datetime(timeline_bucket) + return { + "schema_version": SCHEMA_VERSION, + "source": "moomoo", + "record_type": "option_snapshot", + "capture_id": capture_id, + "time_utc": time_utc, + "captured_at_utc": time_utc, + "time_bucket_utc": time_bucket_utc, + "timeline_bucket_utc": time_bucket_utc, + "timeline_seconds": timeline_seconds, + "market_date": market_date, + "ticker": ticker, + "symbol": ticker, + "owner_code": contract.owner_code, + "expiry": expiry.isoformat(), + "resolved_spot": discovery.spot if discovery is not None else None, + "option_code": contract.option_code, + "strike": contract.strike, + "right": normalized.option_type.lower(), + "provider_update_time": normalized.snapshot_time, + "raw": _jsonable(raw_record), + "normalized": _jsonable(normalized.as_dict()), + } + + +def _records_path(output_dir: Path, source: str, family: str, market_date: str, symbol: str) -> Path: + return output_dir / source / family / f"date={market_date}" / f"ticker={_canonical_ticker(symbol)}" / "records.jsonl" + + +def _batch_records_path(output_dir: Path, source: str, market_date: str, ticker: str) -> Path: + return output_dir / source / "batches" / f"date={market_date}" / f"ticker={_canonical_ticker(ticker)}" / "records.jsonl" + + +def _captures_path(output_dir: Path, market_date: str) -> Path: + return output_dir / "moomoo" / "captures" / f"date={market_date}" / "captures.jsonl" + + +def _append_jsonl(path: Path, record: dict[str, object]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record, sort_keys=True, separators=(",", ":"), default=_json_default)) + handle.write("\n") + + +def _batch_contract_record(option_record: dict[str, object]) -> dict[str, object]: + return { + "option_code": option_record["option_code"], + "strike": option_record["strike"], + "right": option_record["right"], + "provider_update_time": option_record["provider_update_time"], + "raw": option_record["raw"], + "normalized": option_record["normalized"], + } + + +def _canonical_ticker(value: str) -> str: + return value.strip().upper() + + +def _capture_id(value: datetime) -> str: + return value.strftime("%Y%m%dT%H%M%S.%fZ") + + +def _floor_datetime(value: datetime, seconds: int) -> datetime: + epoch_seconds = int(value.timestamp()) + return datetime.fromtimestamp(epoch_seconds - (epoch_seconds % seconds), tz=UTC) + + +def _aware_utc(value: datetime) -> datetime: + if value.tzinfo is None: + value = value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def _format_datetime(value: datetime) -> str: + if value.tzinfo is None: + value = value.replace(tzinfo=UTC) + return value.astimezone(UTC).isoformat().replace("+00:00", "Z") + + +def _jsonable(value: object) -> object: + if value is None or isinstance(value, str | int | bool): + return value + if isinstance(value, float): + return value if math.isfinite(value) else None + if isinstance(value, datetime): + return _format_datetime(value) + if isinstance(value, date): + return value.isoformat() + if isinstance(value, dict): + return {str(key): _jsonable(item) for key, item in value.items()} + if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray): + return [_jsonable(item) for item in value] + if hasattr(value, "item"): + try: + return _jsonable(value.item()) + except Exception: + pass + return str(value) + + +def _json_default(value: object) -> object: + return _jsonable(value) + + +def _parse_date(raw_value: str) -> date: + return date.fromisoformat(raw_value) + + +def _parse_datetime(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + return _aware_utc(parsed) + + +def _next_weekday(value: date) -> date: + while value.weekday() >= 5: + value += timedelta(days=1) + return value + + +def _normalize_max_loops(value: int) -> int | None: + if value < 0: + raise ValueError("--max-loops must be greater than or equal to 0") + if value == 0: + return None + return value + + +def _create_real_client(host: str, port: int) -> MoomooQuoteClient: + try: + from moomoo import OpenQuoteContext + except ImportError as exc: + raise RuntimeError("moomoo-api package is not installed") from exc + return OpenQuoteContext(host=host, port=port) + + +def _print_market_status(status: str, session: MarketSessionWindow, now: datetime) -> None: + print( + json.dumps( + { + "status": status, + "now_utc": _format_datetime(now), + "market_date": session.market_date, + "opens_at_utc": session.opens_at_utc, + "closes_at_utc": session.closes_at_utc, + }, + separators=(",", ":"), + sort_keys=True, + ), + flush=True, + ) + + +if __name__ == "__main__": + main() + + +__all__ = [ + "DEFAULT_RESEARCH_OUTPUT_DIR", + "DEFAULT_RESEARCH_INTERVAL_SECONDS", + "MarketSessionWindow", + "ResearchCaptureSummary", + "capture_research_snapshot_once", + "next_regular_session_window", + "run_market_hours_recorder_loop", + "run_research_recorder_loop", + "main", +] diff --git a/services/collector/tests/test_moomoo_research_recorder.py b/services/collector/tests/test_moomoo_research_recorder.py new file mode 100644 index 0000000..b06da80 --- /dev/null +++ b/services/collector/tests/test_moomoo_research_recorder.py @@ -0,0 +1,259 @@ +from __future__ import annotations + +import json +from datetime import UTC, date, datetime +from pathlib import Path +from zoneinfo import ZoneInfo + +from gammascope_collector.moomoo_config import MoomooCollectorConfig, MoomooSymbolConfig +from gammascope_collector.moomoo_research_recorder import ( + capture_research_snapshot_once, + next_regular_session_window, + run_market_hours_recorder_loop, +) + + +class FakeQuoteClient: + def __init__( + self, + *, + chains: dict[str, list[dict[str, object]]] | None = None, + snapshots: dict[str, dict[str, object]] | None = None, + ) -> None: + self.chains = chains or {} + self.snapshots = snapshots or {} + self.snapshot_calls: list[list[str]] = [] + self.option_chain_calls: list[tuple[str, str, str]] = [] + + def query_subscription(self, is_all_conn: bool = True) -> tuple[int, dict[str, object]]: + return 0, {"is_all_conn": is_all_conn, "sub_list": ["US.SPY"]} + + def get_option_chain(self, code: str, *, start: str, end: str) -> tuple[int, list[dict[str, object]]]: + self.option_chain_calls.append((code, start, end)) + return 0, self.chains.get(code, []) + + def get_market_snapshot(self, code_list: list[str]) -> tuple[int, list[dict[str, object]]]: + self.snapshot_calls.append(list(code_list)) + return 0, [self.snapshots[code] for code in code_list if code in self.snapshots] + + def close(self) -> None: + return None + + +def test_capture_research_snapshot_once_writes_full_raw_option_fields(tmp_path: Path) -> None: + client = FakeQuoteClient( + chains={ + "US.SPY": [ + _option("US.SPY260518C00500000", strike=500, option_type="CALL"), + _option("US.SPY260518P00500000", strike=500, option_type="PUT"), + ] + }, + snapshots={ + "US.SPY": { + "code": "US.SPY", + "last_price": 500.0, + "sec_status": "NORMAL", + }, + "US.SPY260518C00500000": { + "code": "US.SPY260518C00500000", + "name": "SPY 500C", + "last_price": 1.25, + "bid_price": 1.2, + "ask_price": 1.3, + "bid_vol": 11, + "ask_vol": 12, + "volume": 100, + "option_open_interest": 200, + "option_implied_volatility": 30.0, + "option_delta": 0.51, + "option_gamma": 0.02, + "option_vega": 0.15, + "option_theta": -0.04, + "option_rho": 0.03, + "option_premium": 125.0, + "option_net_open_interest": 7, + "provider_extra_field": "keep-me", + "update_time": "2026-05-18 09:31:00", + }, + "US.SPY260518P00500000": { + "code": "US.SPY260518P00500000", + "name": "SPY 500P", + "last_price": 1.35, + "bid_price": 1.3, + "ask_price": 1.4, + "option_open_interest": 220, + "update_time": "2026-05-18 09:31:00", + }, + }, + ) + config = MoomooCollectorConfig( + universe=[ + MoomooSymbolConfig( + symbol="SPY", + owner_code="US.SPY", + strike_window_down=0, + strike_window_up=0, + manual_spot=500, + ) + ] + ) + + summary = capture_research_snapshot_once( + client, + config, + output_dir=tmp_path, + expiry=date(2026, 5, 18), + timeline_seconds=10, + ) + + assert summary.option_rows_written == 2 + assert summary.batch_rows_written == 1 + assert summary.underlying_rows_written == 1 + assert summary.selected_contracts == {"SPY": 2} + assert summary.option_files == [str(tmp_path / "moomoo/options/date=2026-05-18/ticker=SPY/records.jsonl")] + assert summary.batch_files == [str(tmp_path / "moomoo/batches/date=2026-05-18/ticker=SPY/records.jsonl")] + + option_records = _read_jsonl(Path(summary.option_files[0])) + assert option_records[0]["source"] == "moomoo" + assert option_records[0]["record_type"] == "option_snapshot" + assert option_records[0]["ticker"] == "SPY" + assert option_records[0]["symbol"] == "SPY" + assert option_records[0]["time_utc"] == option_records[0]["captured_at_utc"] + assert option_records[0]["time_bucket_utc"] == option_records[0]["timeline_bucket_utc"] + assert option_records[0]["timeline_bucket_utc"].endswith("Z") + assert option_records[0]["timeline_seconds"] == 10 + assert option_records[0]["raw"]["provider_extra_field"] == "keep-me" + assert option_records[0]["raw"]["option_net_open_interest"] == 7 + assert option_records[0]["raw"]["option_premium"] == 125.0 + assert option_records[0]["normalized"]["implied_volatility"] == 0.3 + assert option_records[0]["normalized"]["rho"] == 0.03 + + underlying_records = _read_jsonl(Path(summary.underlying_files[0])) + assert underlying_records[0]["record_type"] == "underlying_snapshot" + assert underlying_records[0]["ticker"] == "SPY" + assert underlying_records[0]["symbol"] == "SPY" + assert underlying_records[0]["provider_code"] == "US.SPY" + assert underlying_records[0]["time_utc"] == underlying_records[0]["captured_at_utc"] + assert underlying_records[0]["timeline_seconds"] == 10 + assert underlying_records[0]["raw"]["sec_status"] == "NORMAL" + + batch_records = _read_jsonl(Path(summary.batch_files[0])) + assert batch_records[0]["record_type"] == "option_snapshot_batch" + assert batch_records[0]["ticker"] == "SPY" + assert batch_records[0]["symbol"] == "SPY" + assert batch_records[0]["time_utc"] == batch_records[0]["captured_at_utc"] + assert batch_records[0]["time_bucket_utc"] == batch_records[0]["timeline_bucket_utc"] + assert batch_records[0]["contract_count"] == 2 + assert batch_records[0]["contracts"][0]["option_code"] == "US.SPY260518C00500000" + assert batch_records[0]["contracts"][0]["raw"]["provider_extra_field"] == "keep-me" + assert batch_records[0]["contracts"][0]["raw"]["option_net_open_interest"] == 7 + assert batch_records[0]["contracts"][0]["normalized"]["implied_volatility"] == 0.3 + + capture_records = _read_jsonl(tmp_path / "moomoo/captures/date=2026-05-18/captures.jsonl") + assert capture_records[0]["time_utc"] == capture_records[0]["captured_at_utc"] + assert capture_records[0]["time_bucket_utc"] == capture_records[0]["timeline_bucket_utc"] + assert capture_records[0]["option_rows_written"] == 2 + assert capture_records[0]["batch_rows_written"] == 1 + assert capture_records[0]["underlying_rows_written"] == 1 + assert capture_records[0]["tickers"] == ["SPY"] + + +def test_capture_research_snapshot_once_records_missing_manual_spot_warning(tmp_path: Path) -> None: + client = FakeQuoteClient() + config = MoomooCollectorConfig( + universe=[ + MoomooSymbolConfig( + symbol="RUT", + owner_code="US..RUT", + strike_window_down=1, + strike_window_up=1, + requires_manual_spot=True, + ) + ] + ) + + summary = capture_research_snapshot_once( + client, + config, + output_dir=tmp_path, + expiry=date(2026, 5, 18), + timeline_seconds=10, + ) + + assert summary.option_rows_written == 0 + assert summary.selected_contracts == {"RUT": 0} + assert summary.warnings == ["RUT requires manual spot and none was supplied"] + assert client.option_chain_calls == [] + + +def test_next_regular_session_window_uses_regular_eastern_market_hours() -> None: + eastern = ZoneInfo("America/New_York") + + session = next_regular_session_window(datetime(2026, 5, 18, 9, 29, tzinfo=eastern)) + + assert session.market_date == "2026-05-18" + assert session.opens_at_utc == "2026-05-18T13:30:00Z" + assert session.closes_at_utc == "2026-05-18T20:00:00Z" + + +def test_next_regular_session_window_rolls_after_close_and_skips_weekends() -> None: + eastern = ZoneInfo("America/New_York") + + monday = next_regular_session_window(datetime(2026, 5, 15, 16, 1, tzinfo=eastern)) + + assert monday.market_date == "2026-05-18" + assert monday.opens_at_utc == "2026-05-18T13:30:00Z" + + +def test_market_hours_loop_captures_when_inside_regular_session(tmp_path: Path) -> None: + client = FakeQuoteClient( + chains={ + "US.SPY": [ + _option("US.SPY260518C00500000", strike=500, option_type="CALL"), + _option("US.SPY260518P00500000", strike=500, option_type="PUT"), + ] + }, + snapshots={ + "US.SPY": {"code": "US.SPY", "last_price": 500.0}, + "US.SPY260518C00500000": {"code": "US.SPY260518C00500000", "last_price": 1.25}, + "US.SPY260518P00500000": {"code": "US.SPY260518P00500000", "last_price": 1.35}, + }, + ) + config = MoomooCollectorConfig( + universe=[ + MoomooSymbolConfig( + symbol="SPY", + owner_code="US.SPY", + strike_window_down=0, + strike_window_up=0, + manual_spot=500, + ) + ] + ) + + summary = run_market_hours_recorder_loop( + client, + config, + output_dir=tmp_path, + expiry=date(2026, 5, 18), + max_loops=1, + now_provider=lambda: datetime(2026, 5, 18, 13, 31, tzinfo=UTC), + sleep=lambda _: None, + ) + + assert summary.option_rows_written == 2 + assert Path(summary.option_files[0]).exists() + + +def _option(code: str, *, strike: float, option_type: str) -> dict[str, object]: + return { + "code": code, + "name": code, + "strike_price": strike, + "option_type": option_type, + "strike_time": "2026-05-18", + } + + +def _read_jsonl(path: Path) -> list[dict[str, object]]: + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] diff --git a/tests/package-scripts.test.mjs b/tests/package-scripts.test.mjs index 2823b88..a57c780 100644 --- a/tests/package-scripts.test.mjs +++ b/tests/package-scripts.test.mjs @@ -19,3 +19,17 @@ test("collector:moomoo-snapshot runs the Moomoo collector from the project virtu "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.moomoo_snapshot", ); }); + +test("collector:moomoo-research-record runs the local ML recorder from the project virtualenv", () => { + assert.equal( + packageJson.scripts["collector:moomoo-research-record"], + "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.moomoo_research_recorder", + ); +}); + +test("collector:moomoo-research-market runs the repeat market-hours recorder", () => { + assert.equal( + packageJson.scripts["collector:moomoo-research-market"], + "PYTHONPATH=services/collector:apps/api .venv/bin/python -m gammascope_collector.moomoo_research_recorder --market-hours --repeat-daily", + ); +});