From 612cd5955146439df46df3e972861bcd4aa77250 Mon Sep 17 00:00:00 2001 From: mrcha033 Date: Thu, 30 Jul 2026 10:10:00 +0900 Subject: [PATCH] feat: add selected opening-hour walk-forward research --- .claude-plugin/marketplace.json | 4 +- .github/workflows/validate.yml | 4 + .../.claude-plugin/plugin.json | 4 +- .../.codex-plugin/plugin.json | 10 +- .../skills/quant-stock-technical/SKILL.md | 10 +- .../quant-stock-technical/agents/openai.yaml | 4 +- .../scripts/analyze_stock.py | 27 +- .../scripts/selected_open1h_outcomes.py | 535 +++++++++++ .../scripts/walk_forward_open1h.py | 854 ++++++++++++++++++ release/catalog.json | 6 +- release/release-notes.md | 8 + skills/quant-stock-technical/SKILL.md | 10 +- .../quant-stock-technical/agents/openai.yaml | 4 +- .../scripts/analyze_stock.py | 27 +- .../scripts/selected_open1h_outcomes.py | 535 +++++++++++ .../scripts/walk_forward_open1h.py | 854 ++++++++++++++++++ tests/test_quant_execution.py | 8 + 17 files changed, 2884 insertions(+), 20 deletions(-) create mode 100644 plugins/quant-stock-technical/skills/quant-stock-technical/scripts/selected_open1h_outcomes.py create mode 100644 plugins/quant-stock-technical/skills/quant-stock-technical/scripts/walk_forward_open1h.py create mode 100644 skills/quant-stock-technical/scripts/selected_open1h_outcomes.py create mode 100644 skills/quant-stock-technical/scripts/walk_forward_open1h.py diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index eec2739..81b744a 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -28,8 +28,8 @@ "name": "quant-stock-technical", "displayName": "Quant Stock Technical", "source": "./plugins/quant-stock-technical", - "description": "Acquire audited KIS Korean and U.S. EOD data and screen a deterministic KOSPI, KOSDAQ, NYSE, and NASDAQ technical universe.", - "version": "0.4.5", + "description": "Acquire audited KIS EOD data, screen four exchanges, and evaluate selected opening-hour outcomes.", + "version": "0.4.6", "author": { "name": "mrcha033" }, diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index f786353..30568e8 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -34,6 +34,10 @@ jobs: run: python -B skills/quant-stock-technical/scripts/collect_open1h_research.py --self-test - name: Run opening-hour evaluator self-test run: python -B skills/quant-stock-technical/scripts/evaluate_open1h_research.py --self-test + - name: Run selected opening-hour outcomes self-test + run: python -B skills/quant-stock-technical/scripts/selected_open1h_outcomes.py --self-test + - name: Run selected opening-hour walk-forward self-test + run: python -B skills/quant-stock-technical/scripts/walk_forward_open1h.py --self-test - name: Run four-exchange universe builder self-test run: python -B skills/quant-stock-technical/scripts/build_universe_manifest.py --self-test - name: Run deterministic universe screen self-test diff --git a/plugins/quant-stock-technical/.claude-plugin/plugin.json b/plugins/quant-stock-technical/.claude-plugin/plugin.json index 2d5f36b..7287cc0 100644 --- a/plugins/quant-stock-technical/.claude-plugin/plugin.json +++ b/plugins/quant-stock-technical/.claude-plugin/plugin.json @@ -2,8 +2,8 @@ "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json", "name": "quant-stock-technical", "displayName": "Quant Stock Technical", - "version": "0.4.5", - "description": "Acquire audited KIS Korean and U.S. EOD data and screen a deterministic KOSPI, KOSDAQ, NYSE, and NASDAQ technical universe.", + "version": "0.4.6", + "description": "Acquire audited KIS EOD data, screen four exchanges, and evaluate selected opening-hour outcomes.", "author": { "name": "mrcha033", "url": "https://github.com/mrcha033" diff --git a/plugins/quant-stock-technical/.codex-plugin/plugin.json b/plugins/quant-stock-technical/.codex-plugin/plugin.json index f6f19bb..4328d9e 100644 --- a/plugins/quant-stock-technical/.codex-plugin/plugin.json +++ b/plugins/quant-stock-technical/.codex-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "quant-stock-technical", - "version": "0.4.5", - "description": "Acquire audited KIS Korean and U.S. EOD data and screen a deterministic four-exchange stock universe.", + "version": "0.4.6", + "description": "Acquire audited KIS EOD data, screen four exchanges, and evaluate selected opening-hour outcomes.", "author": { "name": "mrcha033", "url": "https://github.com/mrcha033" @@ -16,8 +16,8 @@ "skills": "./skills/", "interface": { "displayName": "Quant Stock Technical", - "shortDescription": "Four-exchange EOD acquisition and screening", - "longDescription": "Acquire resumable read-only KIS Korean and U.S. adjusted EOD data, audit invalid provider rows without interpolation, and build a source-hashed KOSPI, KOSDAQ, NYSE, and NASDAQ universe.", + "shortDescription": "Four-exchange screening and opening-hour research", + "longDescription": "Acquire resumable read-only KIS Korean and U.S. adjusted EOD data, build a source-hashed four-exchange universe, and evaluate frozen selected candidates against opening-hour outcomes.", "developerName": "mrcha033", "category": "Finance", "capabilities": [ @@ -28,7 +28,7 @@ "defaultPrompt": [ "Acquire resumable read-only KIS Korean and U.S. adjusted-EOD bundles.", "Calculate a technical report from these completed OHLCV files.", - "Build and screen a frozen four-exchange technical universe." + "Evaluate frozen selected candidates against opening-hour outcomes." ] } } diff --git a/plugins/quant-stock-technical/skills/quant-stock-technical/SKILL.md b/plugins/quant-stock-technical/skills/quant-stock-technical/SKILL.md index 8f312cb..609588b 100644 --- a/plugins/quant-stock-technical/skills/quant-stock-technical/SKILL.md +++ b/plugins/quant-stock-technical/skills/quant-stock-technical/SKILL.md @@ -1,6 +1,6 @@ --- name: quant-stock-technical -description: Acquire resumable read-only KIS Korean and U.S. adjusted-EOD bundles, calculate reproducible numbers-only stock technical analysis, and screen deterministic KOSPI, KOSDAQ, NYSE, and NASDAQ target universes. Use when the user needs completed historical OHLCV and benchmarks, an as-of-date market/ticker report, short-, medium-, and long-horizon opinions, a quantitative risk counterpoint, entry/stop/take-profit levels, a 0-100 technical-strength score, or a frozen exchange-aware screen for a separate execution system. Do not use news, fundamentals, analyst views, narrative judgment, intraday execution, or live-order claims. +description: Acquire resumable read-only KIS Korean and U.S. adjusted-EOD bundles, calculate reproducible numbers-only stock technical analysis, screen deterministic KOSPI, KOSDAQ, NYSE, and NASDAQ target universes, and evaluate frozen selected candidates against opening-hour outcomes. Use when the user needs completed historical OHLCV and benchmarks, an as-of-date market/ticker report, short-, medium-, and long-horizon opinions, a quantitative risk counterpoint, entry/stop/take-profit levels, a 0-100 technical-strength score, a frozen exchange-aware screen, or a read-only point-in-time or explicitly survivor-biased walk-forward diagnostic. Do not use news, fundamentals, analyst views, narrative judgment, intraday execution, or live-order claims. --- # Quant Stock Technical @@ -104,9 +104,17 @@ python3 scripts/build_universe_manifest.py --self-test python3 scripts/analyze_stock_v2.py --self-test python3 scripts/collect_open1h_research.py --self-test python3 scripts/evaluate_open1h_research.py --self-test +python3 scripts/selected_open1h_outcomes.py --self-test +python3 scripts/walk_forward_open1h.py --self-test python3 scripts/screen_universe.py --self-test ``` +For a longer selected-candidate diagnostic, use +`scripts/walk_forward_open1h.py`. It holds one source manifest membership +static across prior sessions and therefore labels the result +`STATIC_SOURCE_MANIFEST_SURVIVOR_BIASED`; it is research evidence, not a +point-in-time universe backtest or promotion evidence. + When `$stock-scenario-story` will run next, save the unmodified JSON output to a file. That downstream skill requires `source_skill`, `result_schema`, `method_version`, and the complete calculator payload; do not summarize or reconstruct the handoff manually. ## Output boundaries diff --git a/plugins/quant-stock-technical/skills/quant-stock-technical/agents/openai.yaml b/plugins/quant-stock-technical/skills/quant-stock-technical/agents/openai.yaml index 25b7e69..5602b74 100644 --- a/plugins/quant-stock-technical/skills/quant-stock-technical/agents/openai.yaml +++ b/plugins/quant-stock-technical/skills/quant-stock-technical/agents/openai.yaml @@ -1,4 +1,4 @@ interface: display_name: "Quant Stock Technical" - short_description: "Four-exchange EOD acquisition and screening" - default_prompt: "Use $quant-stock-technical to acquire resumable read-only KIS Korean and U.S. adjusted-EOD bundles, calculate reproducible technical scores, or build and screen a frozen KOSPI, KOSDAQ, NYSE, and NASDAQ universe." + short_description: "Four-exchange screening and opening-hour research" + default_prompt: "Use $quant-stock-technical to acquire read-only KIS EOD data, calculate reproducible technical scores, screen a frozen four-exchange universe, or evaluate selected opening-hour outcomes." diff --git a/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/analyze_stock.py b/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/analyze_stock.py index 1fba418..4a62bc1 100644 --- a/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/analyze_stock.py +++ b/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/analyze_stock.py @@ -215,6 +215,14 @@ def atr_adx(bars: list[Bar], period: int = 14) -> tuple[list[Optional[float]], l minus_di[index] = 100.0 * minus_value / atr_value denominator = plus_di[index] + minus_di[index] dx[index] = 0.0 if denominator == 0 else 100.0 * abs(plus_di[index] - minus_di[index]) / denominator + else: + # A newly listed or previously suspended instrument can begin with + # a flat true-range seed. Wilder's directional movement is zero + # in that case; leaving DX unavailable would prevent ADX from ever + # initializing, even after the instrument starts trading. + plus_di[index] = 0.0 + minus_di[index] = 0.0 + dx[index] = 0.0 first_adx = 2 * period - 1 initial_dx = [value for value in dx[period : first_adx + 1] if value is not None] @@ -459,7 +467,10 @@ def calculate(ticker_bars: list[Bar], benchmark_bars: list[Bar], market: str, ti percentiles: dict[str, float] = {} reference_counts: dict[str, int] = {} for name, series in {**features, **risk_features}.items(): - percentiles[name], reference_counts[name] = percentile_rank(series, index) + try: + percentiles[name], reference_counts[name] = percentile_rank(series, index) + except BlockedError as exc: + raise BlockedError(f"{name}: {exc}") from exc short_weights = { "return_5": 0.25, @@ -600,6 +611,20 @@ def synthetic_bars(length: int, drift: float, phase: float) -> list[Bar]: def self_test() -> None: ticker_bars = synthetic_bars(1100, 0.0006, 0.0) benchmark_bars = synthetic_bars(1100, 0.0003, 1.0) + flat_then_active = [ + Bar( + day=date(2020, 1, 1) + timedelta(days=index), + open=100.0 if index < 20 else 100.0 + index - 20, + high=100.0 if index < 20 else 101.0 + index - 20, + low=100.0 if index < 20 else 99.0 + index - 20, + close=100.0 if index < 20 else 100.5 + index - 20, + volume=0.0 if index < 20 else 1_000.0, + ) + for index in range(80) + ] + _, restarted_adx, _, _ = atr_adx(flat_then_active, 14) + assert restarted_adx[-1] is not None + assert math.isfinite(restarted_adx[-1]) with tempfile.TemporaryDirectory(prefix="qta-self-test-") as directory: paths = [Path(directory) / "ticker.csv", Path(directory) / "benchmark.csv"] for path, bars in zip(paths, (ticker_bars, benchmark_bars)): diff --git a/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/selected_open1h_outcomes.py b/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/selected_open1h_outcomes.py new file mode 100644 index 0000000..49317fb --- /dev/null +++ b/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/selected_open1h_outcomes.py @@ -0,0 +1,535 @@ +#!/usr/bin/env python3 +"""Collect and compare selected candidates from frozen point-in-time screens.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import date +from decimal import Decimal +from pathlib import Path +from typing import Any + +import analyze_stock as qta1 +import collect_open1h_research as intraday +import evaluate_open1h_research as evaluation +import fetch_kis_kr_eod as shared +import screen_universe as screen +import walk_forward_open1h as walk_forward + + +SCHEMA = "qta-selected-open1h-outcomes/v1" + + +class SelectedOutcomesBlockedError(ValueError): + """Raised when selected point-in-time outcomes cannot be verified.""" + + +def absolute_file(value: str, field: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute() or not path.is_file() or path.is_symlink(): + raise SelectedOutcomesBlockedError( + f"{field} must be an absolute regular non-symlink file" + ) + return path + + +def absolute_directory(value: str, field: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + raise SelectedOutcomesBlockedError(f"{field} must be absolute") + if path.exists() and (not path.is_dir() or path.is_symlink()): + raise SelectedOutcomesBlockedError( + f"{field} must be a non-symlink directory" + ) + return path + + +def load_manifest(path: Path) -> dict[str, Any]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + manifest = screen.normalize_manifest_v2(raw) + except (OSError, json.JSONDecodeError, screen.ScreenBlockedError) as exc: + raise SelectedOutcomesBlockedError( + f"manifest validation failed: {exc}" + ) from exc + supplied_hash = raw.get("manifest_hash") + if not isinstance(supplied_hash, str): + raise SelectedOutcomesBlockedError("manifest_hash is required") + return manifest + + +def selected_rows( + screen_value: dict[str, Any], + *, + market: str, +) -> list[dict[str, Any]]: + selected = screen_value.get("selected") + if not isinstance(selected, dict): + raise SelectedOutcomesBlockedError( + "screen.selected must be an exchange object" + ) + exchanges = intraday.MARKET_CONTRACTS[market]["exchanges"] + rows: list[dict[str, Any]] = [] + for exchange in exchanges: + exchange_rows = selected.get(exchange, []) + if not isinstance(exchange_rows, list): + raise SelectedOutcomesBlockedError( + f"screen.selected.{exchange} must be an array" + ) + for row in exchange_rows: + if ( + not isinstance(row, dict) + or not isinstance(row.get("instrument"), dict) + or not isinstance(row.get("qta"), dict) + ): + raise SelectedOutcomesBlockedError( + "selected row is missing instrument or QTA" + ) + if row["instrument"]["market"] != market: + continue + rows.append(row) + rows.sort( + key=lambda item: ( + exchanges.index(item["instrument"]["exchange"]), + int(item["exchange_rank"]), + item["instrument"]["canonical_symbol"], + ) + ) + return rows + + +def cache_path( + output_directory: Path, + *, + session_date: str, + instrument: dict[str, Any], +) -> Path: + return ( + output_directory + / "bars" + / session_date + / instrument["exchange"] + / f"{instrument['canonical_symbol']}.json" + ) + + +def load_or_fetch( + client: shared.KisReadClient, + output_directory: Path, + *, + market: str, + session_date: str, + instrument: dict[str, Any], +) -> tuple[dict[str, Any], bool]: + path = cache_path( + output_directory, + session_date=session_date, + instrument=instrument, + ) + if path.exists(): + try: + artifact = intraday.normalize_bars_artifact( + json.loads(path.read_text(encoding="utf-8")) + ) + except (OSError, json.JSONDecodeError, intraday.ResearchBlockedError) as exc: + raise SelectedOutcomesBlockedError( + f"cached bars are invalid: {path}: {exc}" + ) from exc + if ( + artifact["market"] != market + or artifact["session_date"] != session_date + or artifact["exchange"] != instrument["exchange"] + or artifact["canonical_symbol"] != instrument["canonical_symbol"] + or artifact["broker_symbol"] != instrument["broker_symbol"] + ): + raise SelectedOutcomesBlockedError( + f"cached bars identity mismatch: {path}" + ) + return artifact, True + raw_rows = intraday.fetch_raw_rows( + client, + market=market, + exchange=instrument["exchange"], + broker_symbol=instrument["broker_symbol"], + session_date=session_date, + ) + artifact = intraday.bars_artifact( + instrument=instrument, + market=market, + session_date=session_date, + raw_rows=raw_rows, + ) + shared.atomic_write_json(path, artifact) + return artifact, False + + +def benchmark_open_regime_bps( + instrument: dict[str, Any], + *, + analysis_date: str, + session_date: str, +) -> Decimal | None: + rows = { + row.day: row + for row in qta1.read_csv(instrument["benchmark_csv"]) + } + previous = rows.get(date.fromisoformat(analysis_date)) + current = rows.get(date.fromisoformat(session_date)) + if previous is None or current is None: + return None + return ( + Decimal(str(current.open)) / Decimal(str(previous.close)) - Decimal(1) + ) * Decimal(10000) + + +def compact( + *, + method: str, + analysis_date: str, + session_date: str, + selected: dict[str, Any], + artifact: dict[str, Any] | None, + bars_reason: str | None, +) -> dict[str, Any]: + instrument = selected["instrument"] + qta = selected["qta"] + regime = benchmark_open_regime_bps( + instrument, + analysis_date=analysis_date, + session_date=session_date, + ) + base = { + "method_version": method, + "analysis_date": analysis_date, + "session_date": session_date, + "exchange": instrument["exchange"], + "canonical_symbol": instrument["canonical_symbol"], + "broker_symbol": instrument["broker_symbol"], + "exchange_rank": selected["exchange_rank"], + "qta_total_score": walk_forward.decimal_text( + walk_forward.decimal_value(qta["total_score"], "total_score") + ), + "entry_price": format( + walk_forward.decimal_value(qta["entry_price"], "entry_price"), + "f", + ), + "stop_price": format( + walk_forward.decimal_value(qta["stop_price"], "stop_price"), + "f", + ), + "take_profit_price": format( + walk_forward.decimal_value( + qta["take_profit_price"], "take_profit_price" + ), + "f", + ), + "market_open_regime_bps": ( + walk_forward.decimal_text(regime) if regime is not None else None + ), + "market_open_regime_pass": ( + regime >= 0 if regime is not None else None + ), + "bars_status": "READY" if artifact is not None else "BLOCKED", + "bars_reason": bars_reason, + } + if artifact is None: + return { + **base, + "coverage_status": None, + "bar_count": 0, + "return_60m_bps": None, + "entry_touched_by_minute_high": None, + "entry_crossed_by_bar_close": None, + "entry_cross_bar_time": None, + "end_return_from_entry_bps": None, + "bars_hash": None, + } + bars = artifact["bars"] + summary = artifact["summary"] + entry = walk_forward.decimal_value(qta["entry_price"], "entry_price") + cross_time = evaluation.first_bar_close_cross(bars, entry) + return { + **base, + "coverage_status": intraday.coverage_summary( + bars, market=artifact["market"] + )["coverage_status"], + "bar_count": len(bars), + "return_60m_bps": summary["return_60m_bps"], + "entry_touched_by_minute_high": ( + Decimal(summary["window_high"]) >= entry + ), + "entry_crossed_by_bar_close": cross_time is not None, + "entry_cross_bar_time": cross_time, + "end_return_from_entry_bps": ( + intraday.bps(Decimal(summary["end_close"]), entry) + if cross_time is not None + else None + ), + "bars_hash": artifact["bars_hash"], + } + + +def run(args: argparse.Namespace) -> dict[str, Any]: + manifest_path = absolute_file(args.manifest, "manifest") + output_directory = absolute_directory( + args.output_directory, "output_directory" + ) + output_directory.mkdir(parents=True, exist_ok=True) + manifest = load_manifest(manifest_path) + if manifest["as_of"] != args.session_date: + raise SelectedOutcomesBlockedError( + "manifest.as_of must equal session_date" + ) + screens: dict[str, dict[str, Any]] = {} + selections: dict[str, list[dict[str, Any]]] = {} + for path_text in args.screen: + path = absolute_file(path_text, "screen") + value = evaluation.normalized_screen(path) + method = value["method_version"] + if method in screens: + raise SelectedOutcomesBlockedError( + f"duplicate method screen: {method}" + ) + if value["manifest_hash"] != manifest["manifest_hash"]: + raise SelectedOutcomesBlockedError( + f"{method} screen manifest hash mismatch" + ) + if value["analysis_date"] != manifest["analysis_date"]: + raise SelectedOutcomesBlockedError( + f"{method} screen analysis date mismatch" + ) + screens[method] = value + selections[method] = selected_rows(value, market=args.market) + if not screens: + raise SelectedOutcomesBlockedError("at least one screen is required") + selected_by_key: dict[tuple[str, str], dict[str, Any]] = {} + for method_rows in selections.values(): + for row in method_rows: + instrument = row["instrument"] + selected_by_key[ + (instrument["exchange"], instrument["canonical_symbol"]) + ] = instrument + key, secret, token = shared.load_api_credentials(args.environment) + client = shared.KisReadClient( + environment=args.environment, + app_key=key, + app_secret=secret, + access_token=token, + interval_ms=args.request_interval_ms, + ) + artifacts: dict[tuple[str, str], dict[str, Any] | None] = {} + errors: dict[tuple[str, str], str | None] = {} + cache_hits = 0 + for key_value, instrument in sorted(selected_by_key.items()): + try: + artifact, cache_hit = load_or_fetch( + client, + output_directory, + market=args.market, + session_date=args.session_date, + instrument=instrument, + ) + artifacts[key_value] = artifact + errors[key_value] = None + cache_hits += int(cache_hit) + except ( + shared.EodBlockedError, + intraday.ResearchBlockedError, + SelectedOutcomesBlockedError, + ) as exc: + artifacts[key_value] = None + errors[key_value] = str(exc) + rows: list[dict[str, Any]] = [] + for method, method_rows in selections.items(): + for selected in method_rows: + instrument = selected["instrument"] + key_value = ( + instrument["exchange"], + instrument["canonical_symbol"], + ) + rows.append( + compact( + method=method, + analysis_date=manifest["analysis_date"], + session_date=args.session_date, + selected=selected, + artifact=artifacts[key_value], + bars_reason=errors[key_value], + ) + ) + methods = sorted({row["method_version"] for row in rows}) + exchanges = intraday.MARKET_CONTRACTS[args.market]["exchanges"] + rows.sort( + key=lambda item: ( + methods.index(item["method_version"]), + exchanges.index(item["exchange"]), + int(item["exchange_rank"]), + item["canonical_symbol"], + ) + ) + summary = { + method: { + exchange: walk_forward.summarize( + [ + row + for row in rows + if row["method_version"] == method + and row["exchange"] == exchange + ] + ) + for exchange in exchanges + } + for method in methods + } + without_hash = { + "schema": SCHEMA, + "source_skill": "quant-stock-technical", + "manifest_path": str(manifest_path), + "manifest_hash": manifest["manifest_hash"], + "analysis_date": manifest["analysis_date"], + "session_date": args.session_date, + "market": args.market, + "methods": methods, + "screen_hashes": { + method: screens[method]["screen_hash"] for method in methods + }, + "selected_union_count": len(selected_by_key), + "summary_by_method_and_exchange": summary, + "candidate_rows": rows, + "request_count": client.request_count, + "retry_count": client.retry_count, + "cache_hit_count": cache_hits, + "api_mutation_count": 0, + "live_enabled": False, + "validation_status": "RESEARCH_ONLY", + } + output = { + **without_hash, + "outcomes_hash": shared.sha256_bytes( + shared.canonical_json(without_hash).encode("utf-8") + ), + } + shared.atomic_write_json(output_directory / "selected-outcomes.json", output) + return output + + +def self_test() -> None: + rows = [ + { + "method_version": "qta-2.0.0", + "session_date": "2026-07-29", + "exchange": "NASDAQ", + "bars_status": "READY", + "return_60m_bps": "50", + "entry_crossed_by_bar_close": True, + "end_return_from_entry_bps": "10", + "market_open_regime_pass": True, + } + ] + summary = walk_forward.summarize(rows) + assert summary["mean_return_60m_bps"] == "50.00" + print( + json.dumps( + { + "schema": SCHEMA, + "self_test": "PASS", + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ) + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--self-test", action="store_true") + parser.add_argument("--manifest") + parser.add_argument("--screen", action="append", default=[]) + parser.add_argument("--market", choices=("KR", "US")) + parser.add_argument("--session-date") + parser.add_argument("--output-directory") + parser.add_argument("--environment", choices=("live", "paper"), default="live") + parser.add_argument("--request-interval-ms", type=int, default=120) + args = parser.parse_args() + if not args.self_test: + missing = [ + name + for name in ( + "manifest", + "market", + "session_date", + "output_directory", + ) + if getattr(args, name) is None + ] + if missing: + parser.error(f"missing required arguments: {', '.join(missing)}") + if not args.screen: + parser.error("--screen is required at least once") + try: + date.fromisoformat(args.session_date) + except ValueError: + parser.error("--session-date must be YYYY-MM-DD") + floor = 1000 if args.environment == "paper" else 100 + if args.request_interval_ms < floor: + parser.error( + f"--request-interval-ms must be at least {floor}" + ) + return args + + +def main() -> int: + args = parse_args() + if args.self_test: + self_test() + return 0 + try: + output = run(args) + except ( + SelectedOutcomesBlockedError, + walk_forward.WalkForwardBlockedError, + shared.EodBlockedError, + intraday.ResearchBlockedError, + OSError, + ValueError, + ) as exc: + print( + json.dumps( + { + "schema": SCHEMA, + "status": "BLOCKED", + "reason": str(exc), + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ), + file=sys.stderr, + ) + return 2 + print( + json.dumps( + { + "schema": SCHEMA, + "status": "READY", + "outcomes_hash": output["outcomes_hash"], + "selected_union_count": output["selected_union_count"], + "request_count": output["request_count"], + "retry_count": output["retry_count"], + "api_mutation_count": 0, + "live_enabled": False, + "output": str( + Path(args.output_directory) / "selected-outcomes.json" + ), + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/walk_forward_open1h.py b/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/walk_forward_open1h.py new file mode 100644 index 0000000..b311ecc --- /dev/null +++ b/plugins/quant-stock-technical/skills/quant-stock-technical/scripts/walk_forward_open1h.py @@ -0,0 +1,854 @@ +#!/usr/bin/env python3 +"""Run a deterministic selected-candidate opening-hour walk-forward study.""" + +from __future__ import annotations + +import argparse +import json +import statistics +import sys +from concurrent.futures import ProcessPoolExecutor +from datetime import date +from decimal import Decimal, InvalidOperation, ROUND_HALF_EVEN +from pathlib import Path +from typing import Any, Iterable + +import analyze_stock as qta1 +import analyze_stock_v2 as qta2 +import collect_open1h_research as intraday +import evaluate_open1h_research as evaluation +import fetch_kis_kr_eod as shared +import screen_universe as screen + + +SCHEMA = "qta-selected-open1h-walk-forward/v1" +UNIVERSE_MODE = "STATIC_SOURCE_MANIFEST_SURVIVOR_BIASED" +METHODS = (qta1.METHOD_VERSION, qta2.METHOD_VERSION) +EXCHANGES = ("KOSPI", "KOSDAQ") + + +class WalkForwardBlockedError(ValueError): + """Raised when a walk-forward study cannot be completed causally.""" + + +def decimal_value(value: Any, field: str) -> Decimal: + if isinstance(value, bool): + raise WalkForwardBlockedError(f"{field} must be decimal") + try: + parsed = Decimal(str(value)) + except (InvalidOperation, ValueError) as exc: + raise WalkForwardBlockedError(f"{field} must be decimal") from exc + if not parsed.is_finite(): + raise WalkForwardBlockedError(f"{field} must be finite") + return parsed + + +def decimal_text(value: Decimal, places: str = "0.01") -> str: + return format( + value.quantize(Decimal(places), rounding=ROUND_HALF_EVEN), + "f", + ) + + +def absolute_file(value: str, field: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute() or not path.is_file() or path.is_symlink(): + raise WalkForwardBlockedError( + f"{field} must be an absolute regular non-symlink file" + ) + return path + + +def absolute_directory(value: str, field: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + raise WalkForwardBlockedError(f"{field} must be absolute") + if path.exists() and (not path.is_dir() or path.is_symlink()): + raise WalkForwardBlockedError( + f"{field} must be a non-symlink directory" + ) + return path + + +def load_manifest(path: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + manifest = screen.normalize_manifest_v2(raw) + except (OSError, json.JSONDecodeError, screen.ScreenBlockedError) as exc: + raise WalkForwardBlockedError( + f"source manifest validation failed: {exc}" + ) from exc + instruments = [ + instrument + for instrument in manifest["instruments"] + if instrument["market"] == "KR" + and instrument["exchange"] in EXCHANGES + ] + if not instruments: + raise WalkForwardBlockedError( + "source manifest has no KOSPI/KOSDAQ instruments" + ) + instruments.sort( + key=lambda item: ( + EXCHANGES.index(item["exchange"]), + item["canonical_symbol"], + ) + ) + for instrument in instruments: + ticker_path = absolute_file( + instrument["ticker_csv"], "instrument.ticker_csv" + ) + benchmark_path = absolute_file( + instrument["benchmark_csv"], "instrument.benchmark_csv" + ) + if shared.sha256_file(ticker_path) != instrument["ticker_csv_sha256"]: + raise WalkForwardBlockedError( + f"ticker hash mismatch: {instrument['canonical_symbol']}" + ) + if ( + shared.sha256_file(benchmark_path) + != instrument["benchmark_csv_sha256"] + ): + raise WalkForwardBlockedError( + f"benchmark hash mismatch: {instrument['benchmark_id']}" + ) + return manifest, instruments + + +def shared_session_pairs( + instruments: list[dict[str, Any]], + *, + end_session: date, + sessions: int, +) -> list[tuple[date, date]]: + benchmark_paths = { + exchange: next( + Path(item["benchmark_csv"]) + for item in instruments + if item["exchange"] == exchange + ) + for exchange in EXCHANGES + } + date_sets = [ + {row.day for row in qta1.read_csv(str(path))} + for path in benchmark_paths.values() + ] + common = sorted( + day for day in set.intersection(*date_sets) if day <= end_session + ) + if len(common) < sessions + 1: + raise WalkForwardBlockedError( + f"benchmarks expose only {max(0, len(common) - 1)} usable sessions" + ) + targets = common[-sessions:] + positions = {day: index for index, day in enumerate(common)} + return [(common[positions[target] - 1], target) for target in targets] + + +def blocked_payload( + instrument: dict[str, Any], + reason: str, + method_version: str, +) -> dict[str, Any]: + return { + "source_skill": "quant-stock-technical", + "result_schema": "quant-stock-technical/v1", + "calculation_status": "BLOCKED", + "reason": reason, + "method_version": method_version, + "market": "KR", + "ticker": instrument["canonical_symbol"], + } + + +def analyze_pair_task( + task: tuple[dict[str, Any], str], +) -> dict[str, Any]: + instrument, analysis_date_text = task + analysis_date = date.fromisoformat(analysis_date_text) + try: + ticker_rows, benchmark_rows = qta1.align( + qta1.read_csv(instrument["ticker_csv"]), + qta1.read_csv(instrument["benchmark_csv"]), + analysis_date, + ) + historical_tick = shared.tick_size( + instrument["exchange"], + Decimal(str(ticker_rows[-1].close)), + ) + common = { + "ticker_bars": ticker_rows, + "benchmark_bars": benchmark_rows, + "market": "KR", + "ticker": instrument["canonical_symbol"], + "tick_size": float(historical_tick), + "source_name": instrument["source_name"], + } + payloads = { + qta1.METHOD_VERSION: qta1.calculate(**common), + qta2.METHOD_VERSION: qta2.calculate(**common), + } + previous_close = Decimal(str(ticker_rows[-1].close)) + except (qta1.BlockedError, OSError, ValueError) as exc: + payloads = { + method: blocked_payload(instrument, str(exc), method) + for method in METHODS + } + previous_close = None + return { + "instrument": instrument, + "previous_close": ( + format(previous_close, "f") if previous_close is not None else None + ), + "payloads": payloads, + } + + +def eligible(payload: dict[str, Any]) -> bool: + return ( + payload.get("calculation_status") == "READY" + and payload.get("setup_status") == "READY" + and decimal_value(payload.get("total_score"), "total_score") + >= Decimal("60") + ) + + +def select_candidates( + analyses: list[dict[str, Any]], + *, + method: str, + top_k: int, +) -> list[dict[str, Any]]: + selected: list[dict[str, Any]] = [] + for exchange in EXCHANGES: + ready = [ + item + for item in analyses + if item["instrument"]["exchange"] == exchange + and eligible(item["payloads"][method]) + ] + ready.sort(key=lambda item: screen.ranking_key(item["payloads"][method])) + for rank, item in enumerate(ready[:top_k], start=1): + selected.append( + { + **item, + "exchange_rank": rank, + } + ) + return selected + + +def bars_cache_path( + output_directory: Path, + *, + session_date: str, + instrument: dict[str, Any], +) -> Path: + return ( + output_directory + / "bars" + / session_date + / instrument["exchange"] + / f"{instrument['canonical_symbol']}.json" + ) + + +def load_or_fetch_bars( + client: shared.KisReadClient, + output_directory: Path, + *, + session_date: str, + instrument: dict[str, Any], +) -> tuple[dict[str, Any], bool]: + path = bars_cache_path( + output_directory, + session_date=session_date, + instrument=instrument, + ) + if path.exists(): + try: + artifact = intraday.normalize_bars_artifact( + json.loads(path.read_text(encoding="utf-8")) + ) + except (OSError, json.JSONDecodeError, intraday.ResearchBlockedError) as exc: + raise WalkForwardBlockedError( + f"invalid cached bars: {path}: {exc}" + ) from exc + if ( + artifact["session_date"] != session_date + or artifact["exchange"] != instrument["exchange"] + or artifact["canonical_symbol"] != instrument["canonical_symbol"] + or artifact["broker_symbol"] != instrument["broker_symbol"] + ): + raise WalkForwardBlockedError( + f"cached bars identity mismatch: {path}" + ) + return artifact, True + raw_rows = intraday.fetch_raw_rows( + client, + market="KR", + exchange=instrument["exchange"], + broker_symbol=instrument["broker_symbol"], + session_date=session_date, + ) + artifact = intraday.bars_artifact( + instrument=instrument, + market="KR", + session_date=session_date, + raw_rows=raw_rows, + ) + shared.atomic_write_json(path, artifact) + return artifact, False + + +def benchmark_rows( + instruments: list[dict[str, Any]], +) -> dict[str, dict[date, qta1.RawBar]]: + output: dict[str, dict[date, qta1.RawBar]] = {} + for exchange in EXCHANGES: + path = next( + item["benchmark_csv"] + for item in instruments + if item["exchange"] == exchange + ) + output[exchange] = {row.day: row for row in qta1.read_csv(path)} + return output + + +def open_regime_bps( + rows: dict[date, qta1.RawBar], + *, + analysis_date: date, + session_date: date, +) -> Decimal: + previous = rows.get(analysis_date) + current = rows.get(session_date) + if previous is None or current is None: + raise WalkForwardBlockedError( + "benchmark is missing an analysis/session pair" + ) + return ( + Decimal(str(current.open)) / Decimal(str(previous.close)) - Decimal(1) + ) * Decimal(10000) + + +def compact_candidate( + *, + method: str, + analysis_date: date, + session_date: date, + selected: dict[str, Any], + artifact: dict[str, Any] | None, + bars_error: str | None, + regime_bps: Decimal, +) -> dict[str, Any]: + instrument = selected["instrument"] + qta = selected["payloads"][method] + base = { + "method_version": method, + "analysis_date": analysis_date.isoformat(), + "session_date": session_date.isoformat(), + "exchange": instrument["exchange"], + "canonical_symbol": instrument["canonical_symbol"], + "broker_symbol": instrument["broker_symbol"], + "exchange_rank": selected["exchange_rank"], + "qta_total_score": decimal_text( + decimal_value(qta["total_score"], "qta_total_score") + ), + "qta_short_score": decimal_text( + decimal_value(qta["short"]["score"], "qta_short_score") + ), + "qta_medium_score": decimal_text( + decimal_value(qta["medium"]["score"], "qta_medium_score") + ), + "qta_long_score": decimal_text( + decimal_value(qta["long"]["score"], "qta_long_score") + ), + "qta_risk_score": decimal_text( + decimal_value(qta["risk"]["score"], "qta_risk_score") + ), + "entry_price": format(decimal_value(qta["entry_price"], "entry"), "f"), + "stop_price": format(decimal_value(qta["stop_price"], "stop"), "f"), + "take_profit_price": format( + decimal_value(qta["take_profit_price"], "take_profit"), "f" + ), + "previous_close": selected["previous_close"], + "market_open_regime_bps": decimal_text(regime_bps), + "market_open_regime_pass": regime_bps >= 0, + "bars_status": "READY" if artifact is not None else "BLOCKED", + "bars_reason": bars_error, + } + if artifact is None: + return { + **base, + "coverage_status": None, + "bar_count": 0, + "return_60m_bps": None, + "maximum_excursion_bps": None, + "minimum_excursion_bps": None, + "entry_touched_by_minute_high": None, + "entry_crossed_by_bar_close": None, + "entry_cross_bar_time": None, + "end_return_from_entry_bps": None, + "bars_hash": None, + } + bars = artifact["bars"] + summary = artifact["summary"] + entry = decimal_value(qta["entry_price"], "entry") + cross_time = evaluation.first_bar_close_cross(bars, entry) + end_close = Decimal(summary["end_close"]) + return { + **base, + "coverage_status": intraday.coverage_summary( + bars, market="KR" + )["coverage_status"], + "bar_count": len(bars), + "return_60m_bps": summary["return_60m_bps"], + "maximum_excursion_bps": summary["maximum_excursion_bps"], + "minimum_excursion_bps": summary["minimum_excursion_bps"], + "entry_touched_by_minute_high": ( + Decimal(summary["window_high"]) >= entry + ), + "entry_crossed_by_bar_close": cross_time is not None, + "entry_cross_bar_time": cross_time, + "end_return_from_entry_bps": ( + intraday.bps(end_close, entry) if cross_time is not None else None + ), + "bars_hash": artifact["bars_hash"], + } + + +def mean(values: list[Decimal]) -> str | None: + if not values: + return None + return decimal_text(sum(values, Decimal(0)) / Decimal(len(values))) + + +def median(values: list[Decimal]) -> str | None: + if not values: + return None + return decimal_text(statistics.median(values)) + + +def fraction(numerator: int, denominator: int) -> str | None: + if denominator == 0: + return None + return decimal_text( + Decimal(numerator) / Decimal(denominator), + "0.000001", + ) + + +def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]: + ready = [row for row in rows if row["bars_status"] == "READY"] + returns = [ + decimal_value(row["return_60m_bps"], "return_60m_bps") + for row in ready + ] + crossed = [ + row for row in ready if row["entry_crossed_by_bar_close"] is True + ] + entry_returns = [ + decimal_value(row["end_return_from_entry_bps"], "entry return") + for row in crossed + ] + regime_evaluated = [ + row + for row in ready + if isinstance(row["market_open_regime_pass"], bool) + ] + regime_ready = [ + row for row in ready if row["market_open_regime_pass"] is True + ] + regime_returns = [ + decimal_value(row["return_60m_bps"], "regime return") + for row in regime_ready + ] + regime_crossed = [ + row + for row in crossed + if row["market_open_regime_pass"] is True + ] + regime_entry_returns = [ + decimal_value(row["end_return_from_entry_bps"], "regime entry return") + for row in regime_crossed + ] + session_means: list[Decimal] = [] + for session_date in sorted({row["session_date"] for row in ready}): + values = [ + decimal_value(row["return_60m_bps"], "session return") + for row in ready + if row["session_date"] == session_date + ] + if values: + session_means.append( + sum(values, Decimal(0)) / Decimal(len(values)) + ) + return { + "selected_count": len(rows), + "bars_ready_count": len(ready), + "bars_blocked_count": len(rows) - len(ready), + "mean_return_60m_bps": mean(returns), + "median_return_60m_bps": median(returns), + "positive_fraction": fraction( + sum(value > 0 for value in returns), len(returns) + ), + "entry_cross_count": len(crossed), + "entry_cross_fraction": fraction(len(crossed), len(ready)), + "mean_end_return_from_entry_bps": mean(entry_returns), + "median_end_return_from_entry_bps": median(entry_returns), + "positive_end_return_from_entry_fraction": fraction( + sum(value > 0 for value in entry_returns), + len(entry_returns), + ), + "market_open_regime_evaluated_count": len(regime_evaluated), + "market_open_regime_pass_count": len(regime_ready), + "market_open_regime_pass_fraction": fraction( + len(regime_ready), len(regime_evaluated) + ), + "market_open_regime_pass_mean_return_60m_bps": mean(regime_returns), + "market_open_regime_entry_cross_count": len(regime_crossed), + "market_open_regime_cross_mean_end_return_from_entry_bps": mean( + regime_entry_returns + ), + "mean_equal_weight_session_return_60m_bps": mean(session_means), + } + + +def grouped_summary(rows: list[dict[str, Any]]) -> dict[str, Any]: + return { + method: { + exchange: summarize( + [ + row + for row in rows + if row["method_version"] == method + and row["exchange"] == exchange + ] + ) + for exchange in EXCHANGES + } + for method in METHODS + } + + +def run(args: argparse.Namespace) -> dict[str, Any]: + manifest_path = absolute_file(args.manifest, "manifest") + output_directory = absolute_directory( + args.output_directory, "output_directory" + ) + output_directory.mkdir(parents=True, exist_ok=True) + manifest, instruments = load_manifest(manifest_path) + pairs = shared_session_pairs( + instruments, + end_session=date.fromisoformat(args.end_session), + sessions=args.sessions, + ) + benchmark = benchmark_rows(instruments) + key, secret, token = shared.load_api_credentials(args.environment) + client = shared.KisReadClient( + environment=args.environment, + app_key=key, + app_secret=secret, + access_token=token, + interval_ms=args.request_interval_ms, + ) + workers = min(8, max(1, args.workers)) + candidate_rows: list[dict[str, Any]] = [] + session_receipts: list[dict[str, Any]] = [] + cache_hits = 0 + for analysis_date, session_date in pairs: + tasks = [ + (instrument, analysis_date.isoformat()) + for instrument in instruments + ] + with ProcessPoolExecutor(max_workers=workers) as executor: + analyses = list( + executor.map(analyze_pair_task, tasks, chunksize=16) + ) + selections = { + method: select_candidates( + analyses, + method=method, + top_k=args.top_k, + ) + for method in METHODS + } + selected_by_key: dict[ + tuple[str, str], dict[str, Any] + ] = {} + for method_rows in selections.values(): + for item in method_rows: + instrument = item["instrument"] + selected_by_key[ + (instrument["exchange"], instrument["canonical_symbol"]) + ] = item + artifacts: dict[tuple[str, str], dict[str, Any] | None] = {} + errors: dict[tuple[str, str], str | None] = {} + for key_value, item in sorted(selected_by_key.items()): + try: + artifact, cache_hit = load_or_fetch_bars( + client, + output_directory, + session_date=session_date.isoformat(), + instrument=item["instrument"], + ) + artifacts[key_value] = artifact + errors[key_value] = None + cache_hits += int(cache_hit) + except ( + shared.EodBlockedError, + intraday.ResearchBlockedError, + WalkForwardBlockedError, + ) as exc: + artifacts[key_value] = None + errors[key_value] = str(exc) + for method, method_rows in selections.items(): + for item in method_rows: + instrument = item["instrument"] + key_value = ( + instrument["exchange"], + instrument["canonical_symbol"], + ) + regime = open_regime_bps( + benchmark[instrument["exchange"]], + analysis_date=analysis_date, + session_date=session_date, + ) + candidate_rows.append( + compact_candidate( + method=method, + analysis_date=analysis_date, + session_date=session_date, + selected=item, + artifact=artifacts[key_value], + bars_error=errors[key_value], + regime_bps=regime, + ) + ) + session_receipts.append( + { + "analysis_date": analysis_date.isoformat(), + "session_date": session_date.isoformat(), + "analyzed_instrument_count": len(analyses), + "qta1_blocked_count": sum( + item["payloads"][qta1.METHOD_VERSION].get( + "calculation_status" + ) + != "READY" + for item in analyses + ), + "qta2_blocked_count": sum( + item["payloads"][qta2.METHOD_VERSION].get( + "calculation_status" + ) + != "READY" + for item in analyses + ), + "qta1_selected_count": len( + selections[qta1.METHOD_VERSION] + ), + "qta2_selected_count": len( + selections[qta2.METHOD_VERSION] + ), + "unique_bar_request_candidates": len(selected_by_key), + "bars_blocked_count": sum( + artifact is None for artifact in artifacts.values() + ), + } + ) + print( + json.dumps(session_receipts[-1], sort_keys=True), + file=sys.stderr, + flush=True, + ) + candidate_rows.sort( + key=lambda item: ( + item["session_date"], + METHODS.index(item["method_version"]), + EXCHANGES.index(item["exchange"]), + item["exchange_rank"], + item["canonical_symbol"], + ) + ) + without_hash = { + "schema": SCHEMA, + "source_skill": "quant-stock-technical", + "source_manifest_path": str(manifest_path), + "source_manifest_hash": manifest["manifest_hash"], + "source_manifest_as_of": manifest["as_of"], + "universe_mode": UNIVERSE_MODE, + "universe_limitations": [ + ( + "the source manifest membership is held static across earlier " + "sessions, creating survivorship and membership bias" + ), + ( + "historical bid-ask spreads and exact quote-cycle market-regime " + "states are unavailable" + ), + ( + "market_open_regime_bps is the exact benchmark opening gap, " + "not the runner's later intraday regime observation" + ), + ], + "market": "KR", + "methods": list(METHODS), + "top_k_per_exchange": args.top_k, + "minimum_total_score": "60", + "session_count": len(pairs), + "first_session_date": pairs[0][1].isoformat(), + "last_session_date": pairs[-1][1].isoformat(), + "session_receipts": session_receipts, + "summary_by_method_and_exchange": grouped_summary(candidate_rows), + "candidate_rows": candidate_rows, + "request_count": client.request_count, + "retry_count": client.retry_count, + "cache_hit_count": cache_hits, + "api_mutation_count": 0, + "live_enabled": False, + "validation_status": "RESEARCH_ONLY", + } + output = { + **without_hash, + "walk_forward_hash": shared.sha256_bytes( + shared.canonical_json(without_hash).encode("utf-8") + ), + } + output_path = output_directory / "walk-forward.json" + shared.atomic_write_json(output_path, output) + return output + + +def self_test() -> None: + rows = [ + { + "method_version": qta1.METHOD_VERSION, + "session_date": "2026-01-02", + "exchange": "KOSPI", + "bars_status": "READY", + "return_60m_bps": "100", + "entry_crossed_by_bar_close": True, + "end_return_from_entry_bps": "25", + "market_open_regime_pass": True, + }, + { + "method_version": qta1.METHOD_VERSION, + "session_date": "2026-01-03", + "exchange": "KOSPI", + "bars_status": "READY", + "return_60m_bps": "-50", + "entry_crossed_by_bar_close": False, + "end_return_from_entry_bps": None, + "market_open_regime_pass": False, + }, + ] + summary = summarize(rows) + assert summary["mean_return_60m_bps"] == "25.00" + assert summary["positive_fraction"] == "0.500000" + assert summary["entry_cross_fraction"] == "0.500000" + assert summary["positive_end_return_from_entry_fraction"] == "1.000000" + assert summary["market_open_regime_pass_mean_return_60m_bps"] == "100.00" + assert summary["market_open_regime_entry_cross_count"] == 1 + print( + json.dumps( + { + "schema": SCHEMA, + "self_test": "PASS", + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ) + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--self-test", action="store_true") + parser.add_argument("--manifest") + parser.add_argument("--output-directory") + parser.add_argument("--end-session") + parser.add_argument("--sessions", type=int, default=20) + parser.add_argument("--top-k", type=int, default=2) + parser.add_argument("--environment", choices=("live", "paper"), default="live") + parser.add_argument("--request-interval-ms", type=int, default=120) + parser.add_argument("--workers", type=int, default=8) + args = parser.parse_args() + if not args.self_test: + missing = [ + name + for name in ("manifest", "output_directory", "end_session") + if getattr(args, name) is None + ] + if missing: + parser.error(f"missing required arguments: {', '.join(missing)}") + try: + date.fromisoformat(args.end_session) + except ValueError as exc: + parser.error("--end-session must be YYYY-MM-DD") + if args.sessions < 2 or args.sessions > 252: + parser.error("--sessions must be between 2 and 252") + if args.top_k < 1 or args.top_k > 20: + parser.error("--top-k must be between 1 and 20") + floor = 1000 if args.environment == "paper" else 100 + if args.request_interval_ms < floor: + parser.error( + f"--request-interval-ms must be at least {floor}" + ) + if args.workers < 1 or args.workers > 8: + parser.error("--workers must be between 1 and 8") + return args + + +def main() -> int: + args = parse_args() + if args.self_test: + self_test() + return 0 + try: + output = run(args) + except ( + WalkForwardBlockedError, + shared.EodBlockedError, + intraday.ResearchBlockedError, + OSError, + ValueError, + ) as exc: + print( + json.dumps( + { + "schema": SCHEMA, + "status": "BLOCKED", + "reason": str(exc), + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ), + file=sys.stderr, + ) + return 2 + print( + json.dumps( + { + "schema": SCHEMA, + "status": "READY", + "session_count": output["session_count"], + "walk_forward_hash": output["walk_forward_hash"], + "output": str( + Path(args.output_directory) / "walk-forward.json" + ), + "request_count": output["request_count"], + "retry_count": output["retry_count"], + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/release/catalog.json b/release/catalog.json index 920af67..400b2bf 100644 --- a/release/catalog.json +++ b/release/catalog.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "distributionVersion": "2.4.4", + "distributionVersion": "2.4.5", "releaseTagPrefix": "skills-v", "guaranteeScope": "Package compatibility guarantees archive layout, manifest placement, and byte integrity. Runtime compatibility remains conditional on each skill's declared tools, authentication, network, and data requirements.", "formats": { @@ -116,7 +116,7 @@ }, { "name": "quant-stock-technical", - "version": "0.4.5", + "version": "0.4.6", "runtime": { "codex": "supported", "claudeCode": "supported", @@ -128,7 +128,7 @@ "user-provided finalized OHLCV and benchmark files, or KIS Open API credentials for read-only Korean and U.S. EOD acquisition", "outbound KRX, Nasdaq Trader, KIS, and benchmark source access" ], - "notes": "Adds research-only qta-2.0.0 for the first regular-session hour, full-universe read-only opening-hour collection/evaluation, deterministic parallel screening, and compact stdout receipts. QTA1 remains byte-equivalent as the stable comparison baseline." + "notes": "Adds read-only selected-candidate point-in-time outcomes and an explicitly survivor-biased 2-252-session Korean opening-hour walk-forward diagnostic. A flat initial true-range seed now initializes ADX at zero instead of making it permanently unavailable, and feature failures identify the exact field. QTA2 remains research-only." }, { "name": "stock-scenario-story", diff --git a/release/release-notes.md b/release/release-notes.md index 1d64011..e4a2088 100644 --- a/release/release-notes.md +++ b/release/release-notes.md @@ -1,6 +1,14 @@ This release contains the versioned distribution packages declared in `release/catalog.json`. +- Distribution 2.4.5 updates `quant-stock-technical` to 0.4.6. It adds + read-only selected-candidate point-in-time outcomes and an explicitly + survivor-biased 2-252-session Korean opening-hour walk-forward diagnostic. + A flat initial true-range seed now initializes ADX at zero instead of making + it permanently unavailable, and blocked calculations identify the exact + missing feature. QTA2 remains research-only; the new diagnostics make no + order calls and always report `live_enabled=false`. + - Distribution 2.4.4 updates `quant-stock-technical` to 0.4.5 and `quant-stock-polling-trader` to 0.5.4. It adds research-only QTA2 scoring from a complete 2026-07-28 first-hour universe diagnostic, turnover diff --git a/skills/quant-stock-technical/SKILL.md b/skills/quant-stock-technical/SKILL.md index 8f312cb..609588b 100644 --- a/skills/quant-stock-technical/SKILL.md +++ b/skills/quant-stock-technical/SKILL.md @@ -1,6 +1,6 @@ --- name: quant-stock-technical -description: Acquire resumable read-only KIS Korean and U.S. adjusted-EOD bundles, calculate reproducible numbers-only stock technical analysis, and screen deterministic KOSPI, KOSDAQ, NYSE, and NASDAQ target universes. Use when the user needs completed historical OHLCV and benchmarks, an as-of-date market/ticker report, short-, medium-, and long-horizon opinions, a quantitative risk counterpoint, entry/stop/take-profit levels, a 0-100 technical-strength score, or a frozen exchange-aware screen for a separate execution system. Do not use news, fundamentals, analyst views, narrative judgment, intraday execution, or live-order claims. +description: Acquire resumable read-only KIS Korean and U.S. adjusted-EOD bundles, calculate reproducible numbers-only stock technical analysis, screen deterministic KOSPI, KOSDAQ, NYSE, and NASDAQ target universes, and evaluate frozen selected candidates against opening-hour outcomes. Use when the user needs completed historical OHLCV and benchmarks, an as-of-date market/ticker report, short-, medium-, and long-horizon opinions, a quantitative risk counterpoint, entry/stop/take-profit levels, a 0-100 technical-strength score, a frozen exchange-aware screen, or a read-only point-in-time or explicitly survivor-biased walk-forward diagnostic. Do not use news, fundamentals, analyst views, narrative judgment, intraday execution, or live-order claims. --- # Quant Stock Technical @@ -104,9 +104,17 @@ python3 scripts/build_universe_manifest.py --self-test python3 scripts/analyze_stock_v2.py --self-test python3 scripts/collect_open1h_research.py --self-test python3 scripts/evaluate_open1h_research.py --self-test +python3 scripts/selected_open1h_outcomes.py --self-test +python3 scripts/walk_forward_open1h.py --self-test python3 scripts/screen_universe.py --self-test ``` +For a longer selected-candidate diagnostic, use +`scripts/walk_forward_open1h.py`. It holds one source manifest membership +static across prior sessions and therefore labels the result +`STATIC_SOURCE_MANIFEST_SURVIVOR_BIASED`; it is research evidence, not a +point-in-time universe backtest or promotion evidence. + When `$stock-scenario-story` will run next, save the unmodified JSON output to a file. That downstream skill requires `source_skill`, `result_schema`, `method_version`, and the complete calculator payload; do not summarize or reconstruct the handoff manually. ## Output boundaries diff --git a/skills/quant-stock-technical/agents/openai.yaml b/skills/quant-stock-technical/agents/openai.yaml index 25b7e69..5602b74 100644 --- a/skills/quant-stock-technical/agents/openai.yaml +++ b/skills/quant-stock-technical/agents/openai.yaml @@ -1,4 +1,4 @@ interface: display_name: "Quant Stock Technical" - short_description: "Four-exchange EOD acquisition and screening" - default_prompt: "Use $quant-stock-technical to acquire resumable read-only KIS Korean and U.S. adjusted-EOD bundles, calculate reproducible technical scores, or build and screen a frozen KOSPI, KOSDAQ, NYSE, and NASDAQ universe." + short_description: "Four-exchange screening and opening-hour research" + default_prompt: "Use $quant-stock-technical to acquire read-only KIS EOD data, calculate reproducible technical scores, screen a frozen four-exchange universe, or evaluate selected opening-hour outcomes." diff --git a/skills/quant-stock-technical/scripts/analyze_stock.py b/skills/quant-stock-technical/scripts/analyze_stock.py index 1fba418..4a62bc1 100644 --- a/skills/quant-stock-technical/scripts/analyze_stock.py +++ b/skills/quant-stock-technical/scripts/analyze_stock.py @@ -215,6 +215,14 @@ def atr_adx(bars: list[Bar], period: int = 14) -> tuple[list[Optional[float]], l minus_di[index] = 100.0 * minus_value / atr_value denominator = plus_di[index] + minus_di[index] dx[index] = 0.0 if denominator == 0 else 100.0 * abs(plus_di[index] - minus_di[index]) / denominator + else: + # A newly listed or previously suspended instrument can begin with + # a flat true-range seed. Wilder's directional movement is zero + # in that case; leaving DX unavailable would prevent ADX from ever + # initializing, even after the instrument starts trading. + plus_di[index] = 0.0 + minus_di[index] = 0.0 + dx[index] = 0.0 first_adx = 2 * period - 1 initial_dx = [value for value in dx[period : first_adx + 1] if value is not None] @@ -459,7 +467,10 @@ def calculate(ticker_bars: list[Bar], benchmark_bars: list[Bar], market: str, ti percentiles: dict[str, float] = {} reference_counts: dict[str, int] = {} for name, series in {**features, **risk_features}.items(): - percentiles[name], reference_counts[name] = percentile_rank(series, index) + try: + percentiles[name], reference_counts[name] = percentile_rank(series, index) + except BlockedError as exc: + raise BlockedError(f"{name}: {exc}") from exc short_weights = { "return_5": 0.25, @@ -600,6 +611,20 @@ def synthetic_bars(length: int, drift: float, phase: float) -> list[Bar]: def self_test() -> None: ticker_bars = synthetic_bars(1100, 0.0006, 0.0) benchmark_bars = synthetic_bars(1100, 0.0003, 1.0) + flat_then_active = [ + Bar( + day=date(2020, 1, 1) + timedelta(days=index), + open=100.0 if index < 20 else 100.0 + index - 20, + high=100.0 if index < 20 else 101.0 + index - 20, + low=100.0 if index < 20 else 99.0 + index - 20, + close=100.0 if index < 20 else 100.5 + index - 20, + volume=0.0 if index < 20 else 1_000.0, + ) + for index in range(80) + ] + _, restarted_adx, _, _ = atr_adx(flat_then_active, 14) + assert restarted_adx[-1] is not None + assert math.isfinite(restarted_adx[-1]) with tempfile.TemporaryDirectory(prefix="qta-self-test-") as directory: paths = [Path(directory) / "ticker.csv", Path(directory) / "benchmark.csv"] for path, bars in zip(paths, (ticker_bars, benchmark_bars)): diff --git a/skills/quant-stock-technical/scripts/selected_open1h_outcomes.py b/skills/quant-stock-technical/scripts/selected_open1h_outcomes.py new file mode 100644 index 0000000..49317fb --- /dev/null +++ b/skills/quant-stock-technical/scripts/selected_open1h_outcomes.py @@ -0,0 +1,535 @@ +#!/usr/bin/env python3 +"""Collect and compare selected candidates from frozen point-in-time screens.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import date +from decimal import Decimal +from pathlib import Path +from typing import Any + +import analyze_stock as qta1 +import collect_open1h_research as intraday +import evaluate_open1h_research as evaluation +import fetch_kis_kr_eod as shared +import screen_universe as screen +import walk_forward_open1h as walk_forward + + +SCHEMA = "qta-selected-open1h-outcomes/v1" + + +class SelectedOutcomesBlockedError(ValueError): + """Raised when selected point-in-time outcomes cannot be verified.""" + + +def absolute_file(value: str, field: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute() or not path.is_file() or path.is_symlink(): + raise SelectedOutcomesBlockedError( + f"{field} must be an absolute regular non-symlink file" + ) + return path + + +def absolute_directory(value: str, field: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + raise SelectedOutcomesBlockedError(f"{field} must be absolute") + if path.exists() and (not path.is_dir() or path.is_symlink()): + raise SelectedOutcomesBlockedError( + f"{field} must be a non-symlink directory" + ) + return path + + +def load_manifest(path: Path) -> dict[str, Any]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + manifest = screen.normalize_manifest_v2(raw) + except (OSError, json.JSONDecodeError, screen.ScreenBlockedError) as exc: + raise SelectedOutcomesBlockedError( + f"manifest validation failed: {exc}" + ) from exc + supplied_hash = raw.get("manifest_hash") + if not isinstance(supplied_hash, str): + raise SelectedOutcomesBlockedError("manifest_hash is required") + return manifest + + +def selected_rows( + screen_value: dict[str, Any], + *, + market: str, +) -> list[dict[str, Any]]: + selected = screen_value.get("selected") + if not isinstance(selected, dict): + raise SelectedOutcomesBlockedError( + "screen.selected must be an exchange object" + ) + exchanges = intraday.MARKET_CONTRACTS[market]["exchanges"] + rows: list[dict[str, Any]] = [] + for exchange in exchanges: + exchange_rows = selected.get(exchange, []) + if not isinstance(exchange_rows, list): + raise SelectedOutcomesBlockedError( + f"screen.selected.{exchange} must be an array" + ) + for row in exchange_rows: + if ( + not isinstance(row, dict) + or not isinstance(row.get("instrument"), dict) + or not isinstance(row.get("qta"), dict) + ): + raise SelectedOutcomesBlockedError( + "selected row is missing instrument or QTA" + ) + if row["instrument"]["market"] != market: + continue + rows.append(row) + rows.sort( + key=lambda item: ( + exchanges.index(item["instrument"]["exchange"]), + int(item["exchange_rank"]), + item["instrument"]["canonical_symbol"], + ) + ) + return rows + + +def cache_path( + output_directory: Path, + *, + session_date: str, + instrument: dict[str, Any], +) -> Path: + return ( + output_directory + / "bars" + / session_date + / instrument["exchange"] + / f"{instrument['canonical_symbol']}.json" + ) + + +def load_or_fetch( + client: shared.KisReadClient, + output_directory: Path, + *, + market: str, + session_date: str, + instrument: dict[str, Any], +) -> tuple[dict[str, Any], bool]: + path = cache_path( + output_directory, + session_date=session_date, + instrument=instrument, + ) + if path.exists(): + try: + artifact = intraday.normalize_bars_artifact( + json.loads(path.read_text(encoding="utf-8")) + ) + except (OSError, json.JSONDecodeError, intraday.ResearchBlockedError) as exc: + raise SelectedOutcomesBlockedError( + f"cached bars are invalid: {path}: {exc}" + ) from exc + if ( + artifact["market"] != market + or artifact["session_date"] != session_date + or artifact["exchange"] != instrument["exchange"] + or artifact["canonical_symbol"] != instrument["canonical_symbol"] + or artifact["broker_symbol"] != instrument["broker_symbol"] + ): + raise SelectedOutcomesBlockedError( + f"cached bars identity mismatch: {path}" + ) + return artifact, True + raw_rows = intraday.fetch_raw_rows( + client, + market=market, + exchange=instrument["exchange"], + broker_symbol=instrument["broker_symbol"], + session_date=session_date, + ) + artifact = intraday.bars_artifact( + instrument=instrument, + market=market, + session_date=session_date, + raw_rows=raw_rows, + ) + shared.atomic_write_json(path, artifact) + return artifact, False + + +def benchmark_open_regime_bps( + instrument: dict[str, Any], + *, + analysis_date: str, + session_date: str, +) -> Decimal | None: + rows = { + row.day: row + for row in qta1.read_csv(instrument["benchmark_csv"]) + } + previous = rows.get(date.fromisoformat(analysis_date)) + current = rows.get(date.fromisoformat(session_date)) + if previous is None or current is None: + return None + return ( + Decimal(str(current.open)) / Decimal(str(previous.close)) - Decimal(1) + ) * Decimal(10000) + + +def compact( + *, + method: str, + analysis_date: str, + session_date: str, + selected: dict[str, Any], + artifact: dict[str, Any] | None, + bars_reason: str | None, +) -> dict[str, Any]: + instrument = selected["instrument"] + qta = selected["qta"] + regime = benchmark_open_regime_bps( + instrument, + analysis_date=analysis_date, + session_date=session_date, + ) + base = { + "method_version": method, + "analysis_date": analysis_date, + "session_date": session_date, + "exchange": instrument["exchange"], + "canonical_symbol": instrument["canonical_symbol"], + "broker_symbol": instrument["broker_symbol"], + "exchange_rank": selected["exchange_rank"], + "qta_total_score": walk_forward.decimal_text( + walk_forward.decimal_value(qta["total_score"], "total_score") + ), + "entry_price": format( + walk_forward.decimal_value(qta["entry_price"], "entry_price"), + "f", + ), + "stop_price": format( + walk_forward.decimal_value(qta["stop_price"], "stop_price"), + "f", + ), + "take_profit_price": format( + walk_forward.decimal_value( + qta["take_profit_price"], "take_profit_price" + ), + "f", + ), + "market_open_regime_bps": ( + walk_forward.decimal_text(regime) if regime is not None else None + ), + "market_open_regime_pass": ( + regime >= 0 if regime is not None else None + ), + "bars_status": "READY" if artifact is not None else "BLOCKED", + "bars_reason": bars_reason, + } + if artifact is None: + return { + **base, + "coverage_status": None, + "bar_count": 0, + "return_60m_bps": None, + "entry_touched_by_minute_high": None, + "entry_crossed_by_bar_close": None, + "entry_cross_bar_time": None, + "end_return_from_entry_bps": None, + "bars_hash": None, + } + bars = artifact["bars"] + summary = artifact["summary"] + entry = walk_forward.decimal_value(qta["entry_price"], "entry_price") + cross_time = evaluation.first_bar_close_cross(bars, entry) + return { + **base, + "coverage_status": intraday.coverage_summary( + bars, market=artifact["market"] + )["coverage_status"], + "bar_count": len(bars), + "return_60m_bps": summary["return_60m_bps"], + "entry_touched_by_minute_high": ( + Decimal(summary["window_high"]) >= entry + ), + "entry_crossed_by_bar_close": cross_time is not None, + "entry_cross_bar_time": cross_time, + "end_return_from_entry_bps": ( + intraday.bps(Decimal(summary["end_close"]), entry) + if cross_time is not None + else None + ), + "bars_hash": artifact["bars_hash"], + } + + +def run(args: argparse.Namespace) -> dict[str, Any]: + manifest_path = absolute_file(args.manifest, "manifest") + output_directory = absolute_directory( + args.output_directory, "output_directory" + ) + output_directory.mkdir(parents=True, exist_ok=True) + manifest = load_manifest(manifest_path) + if manifest["as_of"] != args.session_date: + raise SelectedOutcomesBlockedError( + "manifest.as_of must equal session_date" + ) + screens: dict[str, dict[str, Any]] = {} + selections: dict[str, list[dict[str, Any]]] = {} + for path_text in args.screen: + path = absolute_file(path_text, "screen") + value = evaluation.normalized_screen(path) + method = value["method_version"] + if method in screens: + raise SelectedOutcomesBlockedError( + f"duplicate method screen: {method}" + ) + if value["manifest_hash"] != manifest["manifest_hash"]: + raise SelectedOutcomesBlockedError( + f"{method} screen manifest hash mismatch" + ) + if value["analysis_date"] != manifest["analysis_date"]: + raise SelectedOutcomesBlockedError( + f"{method} screen analysis date mismatch" + ) + screens[method] = value + selections[method] = selected_rows(value, market=args.market) + if not screens: + raise SelectedOutcomesBlockedError("at least one screen is required") + selected_by_key: dict[tuple[str, str], dict[str, Any]] = {} + for method_rows in selections.values(): + for row in method_rows: + instrument = row["instrument"] + selected_by_key[ + (instrument["exchange"], instrument["canonical_symbol"]) + ] = instrument + key, secret, token = shared.load_api_credentials(args.environment) + client = shared.KisReadClient( + environment=args.environment, + app_key=key, + app_secret=secret, + access_token=token, + interval_ms=args.request_interval_ms, + ) + artifacts: dict[tuple[str, str], dict[str, Any] | None] = {} + errors: dict[tuple[str, str], str | None] = {} + cache_hits = 0 + for key_value, instrument in sorted(selected_by_key.items()): + try: + artifact, cache_hit = load_or_fetch( + client, + output_directory, + market=args.market, + session_date=args.session_date, + instrument=instrument, + ) + artifacts[key_value] = artifact + errors[key_value] = None + cache_hits += int(cache_hit) + except ( + shared.EodBlockedError, + intraday.ResearchBlockedError, + SelectedOutcomesBlockedError, + ) as exc: + artifacts[key_value] = None + errors[key_value] = str(exc) + rows: list[dict[str, Any]] = [] + for method, method_rows in selections.items(): + for selected in method_rows: + instrument = selected["instrument"] + key_value = ( + instrument["exchange"], + instrument["canonical_symbol"], + ) + rows.append( + compact( + method=method, + analysis_date=manifest["analysis_date"], + session_date=args.session_date, + selected=selected, + artifact=artifacts[key_value], + bars_reason=errors[key_value], + ) + ) + methods = sorted({row["method_version"] for row in rows}) + exchanges = intraday.MARKET_CONTRACTS[args.market]["exchanges"] + rows.sort( + key=lambda item: ( + methods.index(item["method_version"]), + exchanges.index(item["exchange"]), + int(item["exchange_rank"]), + item["canonical_symbol"], + ) + ) + summary = { + method: { + exchange: walk_forward.summarize( + [ + row + for row in rows + if row["method_version"] == method + and row["exchange"] == exchange + ] + ) + for exchange in exchanges + } + for method in methods + } + without_hash = { + "schema": SCHEMA, + "source_skill": "quant-stock-technical", + "manifest_path": str(manifest_path), + "manifest_hash": manifest["manifest_hash"], + "analysis_date": manifest["analysis_date"], + "session_date": args.session_date, + "market": args.market, + "methods": methods, + "screen_hashes": { + method: screens[method]["screen_hash"] for method in methods + }, + "selected_union_count": len(selected_by_key), + "summary_by_method_and_exchange": summary, + "candidate_rows": rows, + "request_count": client.request_count, + "retry_count": client.retry_count, + "cache_hit_count": cache_hits, + "api_mutation_count": 0, + "live_enabled": False, + "validation_status": "RESEARCH_ONLY", + } + output = { + **without_hash, + "outcomes_hash": shared.sha256_bytes( + shared.canonical_json(without_hash).encode("utf-8") + ), + } + shared.atomic_write_json(output_directory / "selected-outcomes.json", output) + return output + + +def self_test() -> None: + rows = [ + { + "method_version": "qta-2.0.0", + "session_date": "2026-07-29", + "exchange": "NASDAQ", + "bars_status": "READY", + "return_60m_bps": "50", + "entry_crossed_by_bar_close": True, + "end_return_from_entry_bps": "10", + "market_open_regime_pass": True, + } + ] + summary = walk_forward.summarize(rows) + assert summary["mean_return_60m_bps"] == "50.00" + print( + json.dumps( + { + "schema": SCHEMA, + "self_test": "PASS", + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ) + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--self-test", action="store_true") + parser.add_argument("--manifest") + parser.add_argument("--screen", action="append", default=[]) + parser.add_argument("--market", choices=("KR", "US")) + parser.add_argument("--session-date") + parser.add_argument("--output-directory") + parser.add_argument("--environment", choices=("live", "paper"), default="live") + parser.add_argument("--request-interval-ms", type=int, default=120) + args = parser.parse_args() + if not args.self_test: + missing = [ + name + for name in ( + "manifest", + "market", + "session_date", + "output_directory", + ) + if getattr(args, name) is None + ] + if missing: + parser.error(f"missing required arguments: {', '.join(missing)}") + if not args.screen: + parser.error("--screen is required at least once") + try: + date.fromisoformat(args.session_date) + except ValueError: + parser.error("--session-date must be YYYY-MM-DD") + floor = 1000 if args.environment == "paper" else 100 + if args.request_interval_ms < floor: + parser.error( + f"--request-interval-ms must be at least {floor}" + ) + return args + + +def main() -> int: + args = parse_args() + if args.self_test: + self_test() + return 0 + try: + output = run(args) + except ( + SelectedOutcomesBlockedError, + walk_forward.WalkForwardBlockedError, + shared.EodBlockedError, + intraday.ResearchBlockedError, + OSError, + ValueError, + ) as exc: + print( + json.dumps( + { + "schema": SCHEMA, + "status": "BLOCKED", + "reason": str(exc), + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ), + file=sys.stderr, + ) + return 2 + print( + json.dumps( + { + "schema": SCHEMA, + "status": "READY", + "outcomes_hash": output["outcomes_hash"], + "selected_union_count": output["selected_union_count"], + "request_count": output["request_count"], + "retry_count": output["retry_count"], + "api_mutation_count": 0, + "live_enabled": False, + "output": str( + Path(args.output_directory) / "selected-outcomes.json" + ), + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/quant-stock-technical/scripts/walk_forward_open1h.py b/skills/quant-stock-technical/scripts/walk_forward_open1h.py new file mode 100644 index 0000000..b311ecc --- /dev/null +++ b/skills/quant-stock-technical/scripts/walk_forward_open1h.py @@ -0,0 +1,854 @@ +#!/usr/bin/env python3 +"""Run a deterministic selected-candidate opening-hour walk-forward study.""" + +from __future__ import annotations + +import argparse +import json +import statistics +import sys +from concurrent.futures import ProcessPoolExecutor +from datetime import date +from decimal import Decimal, InvalidOperation, ROUND_HALF_EVEN +from pathlib import Path +from typing import Any, Iterable + +import analyze_stock as qta1 +import analyze_stock_v2 as qta2 +import collect_open1h_research as intraday +import evaluate_open1h_research as evaluation +import fetch_kis_kr_eod as shared +import screen_universe as screen + + +SCHEMA = "qta-selected-open1h-walk-forward/v1" +UNIVERSE_MODE = "STATIC_SOURCE_MANIFEST_SURVIVOR_BIASED" +METHODS = (qta1.METHOD_VERSION, qta2.METHOD_VERSION) +EXCHANGES = ("KOSPI", "KOSDAQ") + + +class WalkForwardBlockedError(ValueError): + """Raised when a walk-forward study cannot be completed causally.""" + + +def decimal_value(value: Any, field: str) -> Decimal: + if isinstance(value, bool): + raise WalkForwardBlockedError(f"{field} must be decimal") + try: + parsed = Decimal(str(value)) + except (InvalidOperation, ValueError) as exc: + raise WalkForwardBlockedError(f"{field} must be decimal") from exc + if not parsed.is_finite(): + raise WalkForwardBlockedError(f"{field} must be finite") + return parsed + + +def decimal_text(value: Decimal, places: str = "0.01") -> str: + return format( + value.quantize(Decimal(places), rounding=ROUND_HALF_EVEN), + "f", + ) + + +def absolute_file(value: str, field: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute() or not path.is_file() or path.is_symlink(): + raise WalkForwardBlockedError( + f"{field} must be an absolute regular non-symlink file" + ) + return path + + +def absolute_directory(value: str, field: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + raise WalkForwardBlockedError(f"{field} must be absolute") + if path.exists() and (not path.is_dir() or path.is_symlink()): + raise WalkForwardBlockedError( + f"{field} must be a non-symlink directory" + ) + return path + + +def load_manifest(path: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + manifest = screen.normalize_manifest_v2(raw) + except (OSError, json.JSONDecodeError, screen.ScreenBlockedError) as exc: + raise WalkForwardBlockedError( + f"source manifest validation failed: {exc}" + ) from exc + instruments = [ + instrument + for instrument in manifest["instruments"] + if instrument["market"] == "KR" + and instrument["exchange"] in EXCHANGES + ] + if not instruments: + raise WalkForwardBlockedError( + "source manifest has no KOSPI/KOSDAQ instruments" + ) + instruments.sort( + key=lambda item: ( + EXCHANGES.index(item["exchange"]), + item["canonical_symbol"], + ) + ) + for instrument in instruments: + ticker_path = absolute_file( + instrument["ticker_csv"], "instrument.ticker_csv" + ) + benchmark_path = absolute_file( + instrument["benchmark_csv"], "instrument.benchmark_csv" + ) + if shared.sha256_file(ticker_path) != instrument["ticker_csv_sha256"]: + raise WalkForwardBlockedError( + f"ticker hash mismatch: {instrument['canonical_symbol']}" + ) + if ( + shared.sha256_file(benchmark_path) + != instrument["benchmark_csv_sha256"] + ): + raise WalkForwardBlockedError( + f"benchmark hash mismatch: {instrument['benchmark_id']}" + ) + return manifest, instruments + + +def shared_session_pairs( + instruments: list[dict[str, Any]], + *, + end_session: date, + sessions: int, +) -> list[tuple[date, date]]: + benchmark_paths = { + exchange: next( + Path(item["benchmark_csv"]) + for item in instruments + if item["exchange"] == exchange + ) + for exchange in EXCHANGES + } + date_sets = [ + {row.day for row in qta1.read_csv(str(path))} + for path in benchmark_paths.values() + ] + common = sorted( + day for day in set.intersection(*date_sets) if day <= end_session + ) + if len(common) < sessions + 1: + raise WalkForwardBlockedError( + f"benchmarks expose only {max(0, len(common) - 1)} usable sessions" + ) + targets = common[-sessions:] + positions = {day: index for index, day in enumerate(common)} + return [(common[positions[target] - 1], target) for target in targets] + + +def blocked_payload( + instrument: dict[str, Any], + reason: str, + method_version: str, +) -> dict[str, Any]: + return { + "source_skill": "quant-stock-technical", + "result_schema": "quant-stock-technical/v1", + "calculation_status": "BLOCKED", + "reason": reason, + "method_version": method_version, + "market": "KR", + "ticker": instrument["canonical_symbol"], + } + + +def analyze_pair_task( + task: tuple[dict[str, Any], str], +) -> dict[str, Any]: + instrument, analysis_date_text = task + analysis_date = date.fromisoformat(analysis_date_text) + try: + ticker_rows, benchmark_rows = qta1.align( + qta1.read_csv(instrument["ticker_csv"]), + qta1.read_csv(instrument["benchmark_csv"]), + analysis_date, + ) + historical_tick = shared.tick_size( + instrument["exchange"], + Decimal(str(ticker_rows[-1].close)), + ) + common = { + "ticker_bars": ticker_rows, + "benchmark_bars": benchmark_rows, + "market": "KR", + "ticker": instrument["canonical_symbol"], + "tick_size": float(historical_tick), + "source_name": instrument["source_name"], + } + payloads = { + qta1.METHOD_VERSION: qta1.calculate(**common), + qta2.METHOD_VERSION: qta2.calculate(**common), + } + previous_close = Decimal(str(ticker_rows[-1].close)) + except (qta1.BlockedError, OSError, ValueError) as exc: + payloads = { + method: blocked_payload(instrument, str(exc), method) + for method in METHODS + } + previous_close = None + return { + "instrument": instrument, + "previous_close": ( + format(previous_close, "f") if previous_close is not None else None + ), + "payloads": payloads, + } + + +def eligible(payload: dict[str, Any]) -> bool: + return ( + payload.get("calculation_status") == "READY" + and payload.get("setup_status") == "READY" + and decimal_value(payload.get("total_score"), "total_score") + >= Decimal("60") + ) + + +def select_candidates( + analyses: list[dict[str, Any]], + *, + method: str, + top_k: int, +) -> list[dict[str, Any]]: + selected: list[dict[str, Any]] = [] + for exchange in EXCHANGES: + ready = [ + item + for item in analyses + if item["instrument"]["exchange"] == exchange + and eligible(item["payloads"][method]) + ] + ready.sort(key=lambda item: screen.ranking_key(item["payloads"][method])) + for rank, item in enumerate(ready[:top_k], start=1): + selected.append( + { + **item, + "exchange_rank": rank, + } + ) + return selected + + +def bars_cache_path( + output_directory: Path, + *, + session_date: str, + instrument: dict[str, Any], +) -> Path: + return ( + output_directory + / "bars" + / session_date + / instrument["exchange"] + / f"{instrument['canonical_symbol']}.json" + ) + + +def load_or_fetch_bars( + client: shared.KisReadClient, + output_directory: Path, + *, + session_date: str, + instrument: dict[str, Any], +) -> tuple[dict[str, Any], bool]: + path = bars_cache_path( + output_directory, + session_date=session_date, + instrument=instrument, + ) + if path.exists(): + try: + artifact = intraday.normalize_bars_artifact( + json.loads(path.read_text(encoding="utf-8")) + ) + except (OSError, json.JSONDecodeError, intraday.ResearchBlockedError) as exc: + raise WalkForwardBlockedError( + f"invalid cached bars: {path}: {exc}" + ) from exc + if ( + artifact["session_date"] != session_date + or artifact["exchange"] != instrument["exchange"] + or artifact["canonical_symbol"] != instrument["canonical_symbol"] + or artifact["broker_symbol"] != instrument["broker_symbol"] + ): + raise WalkForwardBlockedError( + f"cached bars identity mismatch: {path}" + ) + return artifact, True + raw_rows = intraday.fetch_raw_rows( + client, + market="KR", + exchange=instrument["exchange"], + broker_symbol=instrument["broker_symbol"], + session_date=session_date, + ) + artifact = intraday.bars_artifact( + instrument=instrument, + market="KR", + session_date=session_date, + raw_rows=raw_rows, + ) + shared.atomic_write_json(path, artifact) + return artifact, False + + +def benchmark_rows( + instruments: list[dict[str, Any]], +) -> dict[str, dict[date, qta1.RawBar]]: + output: dict[str, dict[date, qta1.RawBar]] = {} + for exchange in EXCHANGES: + path = next( + item["benchmark_csv"] + for item in instruments + if item["exchange"] == exchange + ) + output[exchange] = {row.day: row for row in qta1.read_csv(path)} + return output + + +def open_regime_bps( + rows: dict[date, qta1.RawBar], + *, + analysis_date: date, + session_date: date, +) -> Decimal: + previous = rows.get(analysis_date) + current = rows.get(session_date) + if previous is None or current is None: + raise WalkForwardBlockedError( + "benchmark is missing an analysis/session pair" + ) + return ( + Decimal(str(current.open)) / Decimal(str(previous.close)) - Decimal(1) + ) * Decimal(10000) + + +def compact_candidate( + *, + method: str, + analysis_date: date, + session_date: date, + selected: dict[str, Any], + artifact: dict[str, Any] | None, + bars_error: str | None, + regime_bps: Decimal, +) -> dict[str, Any]: + instrument = selected["instrument"] + qta = selected["payloads"][method] + base = { + "method_version": method, + "analysis_date": analysis_date.isoformat(), + "session_date": session_date.isoformat(), + "exchange": instrument["exchange"], + "canonical_symbol": instrument["canonical_symbol"], + "broker_symbol": instrument["broker_symbol"], + "exchange_rank": selected["exchange_rank"], + "qta_total_score": decimal_text( + decimal_value(qta["total_score"], "qta_total_score") + ), + "qta_short_score": decimal_text( + decimal_value(qta["short"]["score"], "qta_short_score") + ), + "qta_medium_score": decimal_text( + decimal_value(qta["medium"]["score"], "qta_medium_score") + ), + "qta_long_score": decimal_text( + decimal_value(qta["long"]["score"], "qta_long_score") + ), + "qta_risk_score": decimal_text( + decimal_value(qta["risk"]["score"], "qta_risk_score") + ), + "entry_price": format(decimal_value(qta["entry_price"], "entry"), "f"), + "stop_price": format(decimal_value(qta["stop_price"], "stop"), "f"), + "take_profit_price": format( + decimal_value(qta["take_profit_price"], "take_profit"), "f" + ), + "previous_close": selected["previous_close"], + "market_open_regime_bps": decimal_text(regime_bps), + "market_open_regime_pass": regime_bps >= 0, + "bars_status": "READY" if artifact is not None else "BLOCKED", + "bars_reason": bars_error, + } + if artifact is None: + return { + **base, + "coverage_status": None, + "bar_count": 0, + "return_60m_bps": None, + "maximum_excursion_bps": None, + "minimum_excursion_bps": None, + "entry_touched_by_minute_high": None, + "entry_crossed_by_bar_close": None, + "entry_cross_bar_time": None, + "end_return_from_entry_bps": None, + "bars_hash": None, + } + bars = artifact["bars"] + summary = artifact["summary"] + entry = decimal_value(qta["entry_price"], "entry") + cross_time = evaluation.first_bar_close_cross(bars, entry) + end_close = Decimal(summary["end_close"]) + return { + **base, + "coverage_status": intraday.coverage_summary( + bars, market="KR" + )["coverage_status"], + "bar_count": len(bars), + "return_60m_bps": summary["return_60m_bps"], + "maximum_excursion_bps": summary["maximum_excursion_bps"], + "minimum_excursion_bps": summary["minimum_excursion_bps"], + "entry_touched_by_minute_high": ( + Decimal(summary["window_high"]) >= entry + ), + "entry_crossed_by_bar_close": cross_time is not None, + "entry_cross_bar_time": cross_time, + "end_return_from_entry_bps": ( + intraday.bps(end_close, entry) if cross_time is not None else None + ), + "bars_hash": artifact["bars_hash"], + } + + +def mean(values: list[Decimal]) -> str | None: + if not values: + return None + return decimal_text(sum(values, Decimal(0)) / Decimal(len(values))) + + +def median(values: list[Decimal]) -> str | None: + if not values: + return None + return decimal_text(statistics.median(values)) + + +def fraction(numerator: int, denominator: int) -> str | None: + if denominator == 0: + return None + return decimal_text( + Decimal(numerator) / Decimal(denominator), + "0.000001", + ) + + +def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]: + ready = [row for row in rows if row["bars_status"] == "READY"] + returns = [ + decimal_value(row["return_60m_bps"], "return_60m_bps") + for row in ready + ] + crossed = [ + row for row in ready if row["entry_crossed_by_bar_close"] is True + ] + entry_returns = [ + decimal_value(row["end_return_from_entry_bps"], "entry return") + for row in crossed + ] + regime_evaluated = [ + row + for row in ready + if isinstance(row["market_open_regime_pass"], bool) + ] + regime_ready = [ + row for row in ready if row["market_open_regime_pass"] is True + ] + regime_returns = [ + decimal_value(row["return_60m_bps"], "regime return") + for row in regime_ready + ] + regime_crossed = [ + row + for row in crossed + if row["market_open_regime_pass"] is True + ] + regime_entry_returns = [ + decimal_value(row["end_return_from_entry_bps"], "regime entry return") + for row in regime_crossed + ] + session_means: list[Decimal] = [] + for session_date in sorted({row["session_date"] for row in ready}): + values = [ + decimal_value(row["return_60m_bps"], "session return") + for row in ready + if row["session_date"] == session_date + ] + if values: + session_means.append( + sum(values, Decimal(0)) / Decimal(len(values)) + ) + return { + "selected_count": len(rows), + "bars_ready_count": len(ready), + "bars_blocked_count": len(rows) - len(ready), + "mean_return_60m_bps": mean(returns), + "median_return_60m_bps": median(returns), + "positive_fraction": fraction( + sum(value > 0 for value in returns), len(returns) + ), + "entry_cross_count": len(crossed), + "entry_cross_fraction": fraction(len(crossed), len(ready)), + "mean_end_return_from_entry_bps": mean(entry_returns), + "median_end_return_from_entry_bps": median(entry_returns), + "positive_end_return_from_entry_fraction": fraction( + sum(value > 0 for value in entry_returns), + len(entry_returns), + ), + "market_open_regime_evaluated_count": len(regime_evaluated), + "market_open_regime_pass_count": len(regime_ready), + "market_open_regime_pass_fraction": fraction( + len(regime_ready), len(regime_evaluated) + ), + "market_open_regime_pass_mean_return_60m_bps": mean(regime_returns), + "market_open_regime_entry_cross_count": len(regime_crossed), + "market_open_regime_cross_mean_end_return_from_entry_bps": mean( + regime_entry_returns + ), + "mean_equal_weight_session_return_60m_bps": mean(session_means), + } + + +def grouped_summary(rows: list[dict[str, Any]]) -> dict[str, Any]: + return { + method: { + exchange: summarize( + [ + row + for row in rows + if row["method_version"] == method + and row["exchange"] == exchange + ] + ) + for exchange in EXCHANGES + } + for method in METHODS + } + + +def run(args: argparse.Namespace) -> dict[str, Any]: + manifest_path = absolute_file(args.manifest, "manifest") + output_directory = absolute_directory( + args.output_directory, "output_directory" + ) + output_directory.mkdir(parents=True, exist_ok=True) + manifest, instruments = load_manifest(manifest_path) + pairs = shared_session_pairs( + instruments, + end_session=date.fromisoformat(args.end_session), + sessions=args.sessions, + ) + benchmark = benchmark_rows(instruments) + key, secret, token = shared.load_api_credentials(args.environment) + client = shared.KisReadClient( + environment=args.environment, + app_key=key, + app_secret=secret, + access_token=token, + interval_ms=args.request_interval_ms, + ) + workers = min(8, max(1, args.workers)) + candidate_rows: list[dict[str, Any]] = [] + session_receipts: list[dict[str, Any]] = [] + cache_hits = 0 + for analysis_date, session_date in pairs: + tasks = [ + (instrument, analysis_date.isoformat()) + for instrument in instruments + ] + with ProcessPoolExecutor(max_workers=workers) as executor: + analyses = list( + executor.map(analyze_pair_task, tasks, chunksize=16) + ) + selections = { + method: select_candidates( + analyses, + method=method, + top_k=args.top_k, + ) + for method in METHODS + } + selected_by_key: dict[ + tuple[str, str], dict[str, Any] + ] = {} + for method_rows in selections.values(): + for item in method_rows: + instrument = item["instrument"] + selected_by_key[ + (instrument["exchange"], instrument["canonical_symbol"]) + ] = item + artifacts: dict[tuple[str, str], dict[str, Any] | None] = {} + errors: dict[tuple[str, str], str | None] = {} + for key_value, item in sorted(selected_by_key.items()): + try: + artifact, cache_hit = load_or_fetch_bars( + client, + output_directory, + session_date=session_date.isoformat(), + instrument=item["instrument"], + ) + artifacts[key_value] = artifact + errors[key_value] = None + cache_hits += int(cache_hit) + except ( + shared.EodBlockedError, + intraday.ResearchBlockedError, + WalkForwardBlockedError, + ) as exc: + artifacts[key_value] = None + errors[key_value] = str(exc) + for method, method_rows in selections.items(): + for item in method_rows: + instrument = item["instrument"] + key_value = ( + instrument["exchange"], + instrument["canonical_symbol"], + ) + regime = open_regime_bps( + benchmark[instrument["exchange"]], + analysis_date=analysis_date, + session_date=session_date, + ) + candidate_rows.append( + compact_candidate( + method=method, + analysis_date=analysis_date, + session_date=session_date, + selected=item, + artifact=artifacts[key_value], + bars_error=errors[key_value], + regime_bps=regime, + ) + ) + session_receipts.append( + { + "analysis_date": analysis_date.isoformat(), + "session_date": session_date.isoformat(), + "analyzed_instrument_count": len(analyses), + "qta1_blocked_count": sum( + item["payloads"][qta1.METHOD_VERSION].get( + "calculation_status" + ) + != "READY" + for item in analyses + ), + "qta2_blocked_count": sum( + item["payloads"][qta2.METHOD_VERSION].get( + "calculation_status" + ) + != "READY" + for item in analyses + ), + "qta1_selected_count": len( + selections[qta1.METHOD_VERSION] + ), + "qta2_selected_count": len( + selections[qta2.METHOD_VERSION] + ), + "unique_bar_request_candidates": len(selected_by_key), + "bars_blocked_count": sum( + artifact is None for artifact in artifacts.values() + ), + } + ) + print( + json.dumps(session_receipts[-1], sort_keys=True), + file=sys.stderr, + flush=True, + ) + candidate_rows.sort( + key=lambda item: ( + item["session_date"], + METHODS.index(item["method_version"]), + EXCHANGES.index(item["exchange"]), + item["exchange_rank"], + item["canonical_symbol"], + ) + ) + without_hash = { + "schema": SCHEMA, + "source_skill": "quant-stock-technical", + "source_manifest_path": str(manifest_path), + "source_manifest_hash": manifest["manifest_hash"], + "source_manifest_as_of": manifest["as_of"], + "universe_mode": UNIVERSE_MODE, + "universe_limitations": [ + ( + "the source manifest membership is held static across earlier " + "sessions, creating survivorship and membership bias" + ), + ( + "historical bid-ask spreads and exact quote-cycle market-regime " + "states are unavailable" + ), + ( + "market_open_regime_bps is the exact benchmark opening gap, " + "not the runner's later intraday regime observation" + ), + ], + "market": "KR", + "methods": list(METHODS), + "top_k_per_exchange": args.top_k, + "minimum_total_score": "60", + "session_count": len(pairs), + "first_session_date": pairs[0][1].isoformat(), + "last_session_date": pairs[-1][1].isoformat(), + "session_receipts": session_receipts, + "summary_by_method_and_exchange": grouped_summary(candidate_rows), + "candidate_rows": candidate_rows, + "request_count": client.request_count, + "retry_count": client.retry_count, + "cache_hit_count": cache_hits, + "api_mutation_count": 0, + "live_enabled": False, + "validation_status": "RESEARCH_ONLY", + } + output = { + **without_hash, + "walk_forward_hash": shared.sha256_bytes( + shared.canonical_json(without_hash).encode("utf-8") + ), + } + output_path = output_directory / "walk-forward.json" + shared.atomic_write_json(output_path, output) + return output + + +def self_test() -> None: + rows = [ + { + "method_version": qta1.METHOD_VERSION, + "session_date": "2026-01-02", + "exchange": "KOSPI", + "bars_status": "READY", + "return_60m_bps": "100", + "entry_crossed_by_bar_close": True, + "end_return_from_entry_bps": "25", + "market_open_regime_pass": True, + }, + { + "method_version": qta1.METHOD_VERSION, + "session_date": "2026-01-03", + "exchange": "KOSPI", + "bars_status": "READY", + "return_60m_bps": "-50", + "entry_crossed_by_bar_close": False, + "end_return_from_entry_bps": None, + "market_open_regime_pass": False, + }, + ] + summary = summarize(rows) + assert summary["mean_return_60m_bps"] == "25.00" + assert summary["positive_fraction"] == "0.500000" + assert summary["entry_cross_fraction"] == "0.500000" + assert summary["positive_end_return_from_entry_fraction"] == "1.000000" + assert summary["market_open_regime_pass_mean_return_60m_bps"] == "100.00" + assert summary["market_open_regime_entry_cross_count"] == 1 + print( + json.dumps( + { + "schema": SCHEMA, + "self_test": "PASS", + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ) + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--self-test", action="store_true") + parser.add_argument("--manifest") + parser.add_argument("--output-directory") + parser.add_argument("--end-session") + parser.add_argument("--sessions", type=int, default=20) + parser.add_argument("--top-k", type=int, default=2) + parser.add_argument("--environment", choices=("live", "paper"), default="live") + parser.add_argument("--request-interval-ms", type=int, default=120) + parser.add_argument("--workers", type=int, default=8) + args = parser.parse_args() + if not args.self_test: + missing = [ + name + for name in ("manifest", "output_directory", "end_session") + if getattr(args, name) is None + ] + if missing: + parser.error(f"missing required arguments: {', '.join(missing)}") + try: + date.fromisoformat(args.end_session) + except ValueError as exc: + parser.error("--end-session must be YYYY-MM-DD") + if args.sessions < 2 or args.sessions > 252: + parser.error("--sessions must be between 2 and 252") + if args.top_k < 1 or args.top_k > 20: + parser.error("--top-k must be between 1 and 20") + floor = 1000 if args.environment == "paper" else 100 + if args.request_interval_ms < floor: + parser.error( + f"--request-interval-ms must be at least {floor}" + ) + if args.workers < 1 or args.workers > 8: + parser.error("--workers must be between 1 and 8") + return args + + +def main() -> int: + args = parse_args() + if args.self_test: + self_test() + return 0 + try: + output = run(args) + except ( + WalkForwardBlockedError, + shared.EodBlockedError, + intraday.ResearchBlockedError, + OSError, + ValueError, + ) as exc: + print( + json.dumps( + { + "schema": SCHEMA, + "status": "BLOCKED", + "reason": str(exc), + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ), + file=sys.stderr, + ) + return 2 + print( + json.dumps( + { + "schema": SCHEMA, + "status": "READY", + "session_count": output["session_count"], + "walk_forward_hash": output["walk_forward_hash"], + "output": str( + Path(args.output_directory) / "walk-forward.json" + ), + "request_count": output["request_count"], + "retry_count": output["retry_count"], + "api_mutation_count": 0, + "live_enabled": False, + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_quant_execution.py b/tests/test_quant_execution.py index fd79bc4..c9cabc5 100755 --- a/tests/test_quant_execution.py +++ b/tests/test_quant_execution.py @@ -289,6 +289,14 @@ def main() -> None: "skills/quant-stock-technical/scripts/fetch_kis_us_eod.py", "--self-test", ) + run( + "skills/quant-stock-technical/scripts/selected_open1h_outcomes.py", + "--self-test", + ) + run( + "skills/quant-stock-technical/scripts/walk_forward_open1h.py", + "--self-test", + ) run("skills/quant-stock-technical/scripts/screen_universe.py", "--self-test") run( "skills/quant-stock-polling-trader/scripts/execution_core.py",