scripts/build_model_book.py
#!/usr/bin/env python3
"""
Stockbee Setup Fluency Trainer
Build and maintain a Stockbee-style setup model book from momentum-burst
screener outputs. The trainer records every candidate as a study example, then
updates 3/5-day forward outcomes with MFE/MAE and outcome tags so the trader can
learn which setup variants are working.
Input modes:
1. ingest - add candidates from stockbee-momentum-burst-screener JSON
2. update - attach 3/5-day forward outcomes from offline JSON or FMP
3. summarize - aggregate model-book evidence by rating, trigger, and tags
The script is intentionally signal/research oriented. It does not place orders,
generate broker instructions, or treat screener candidates as automatic trades.
"""
from __future__ import annotations
import argparse
import json
import math
import os
import statistics
import sys
import time
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any
try:
import requests
except ImportError: # pragma: no cover - optional FMP path
requests = None
SCHEMA_VERSION = "1.0"
SKILL_NAME = "stockbee-setup-fluency-trainer"
DEFAULT_HORIZONS = (3, 5)
DEFAULT_MODEL_BOOK = "state/stockbee/model_book.jsonl"
@dataclass(frozen=True)
class Bar:
"""Normalized OHLCV bar sorted oldest -> newest."""
date: str
open: float
high: float
low: float
close: float
volume: int = 0
class ApiCallBudgetExceeded(Exception):
"""Raised when max_api_calls is exhausted."""
class FMPClient:
"""Small /stable-first FMP client for daily OHLCV outcome updates."""
STABLE_URL = "https://financialmodelingprep.com/stable"
V3_URL = "https://financialmodelingprep.com/api/v3"
RATE_LIMIT_DELAY = 0.30
def __init__(self, api_key: str | None = None, max_api_calls: int = 200):
if requests is None:
raise RuntimeError("requests is not installed; use --prices-json or install requests")
self.api_key = api_key or os.getenv("FMP_API_KEY")
if not self.api_key:
raise ValueError("FMP API key required. Set FMP_API_KEY or pass --api-key.")
self.session = requests.Session()
self.session.headers.update({"apikey": self.api_key})
self.max_api_calls = max_api_calls
self.api_calls_made = 0
self.last_call_time = 0.0
self.cache: dict[str, Any] = {}
def _get(self, url: str, params: dict[str, Any] | None = None, quiet: bool = False) -> Any:
if self.api_calls_made >= self.max_api_calls:
raise ApiCallBudgetExceeded(
f"API budget exhausted: {self.api_calls_made}/{self.max_api_calls} calls used"
)
elapsed = time.time() - self.last_call_time
if elapsed < self.RATE_LIMIT_DELAY:
time.sleep(self.RATE_LIMIT_DELAY - elapsed)
response = self.session.get(url, params=params or {}, timeout=30)
self.last_call_time = time.time()
self.api_calls_made += 1
if response.status_code == 200:
return response.json()
if not quiet:
print(
f"ERROR: FMP request failed: HTTP {response.status_code} - {response.text[:200]}",
file=sys.stderr,
)
return None
def get_historical_prices(self, symbol: str, days: int = 80) -> list[dict[str, Any]]:
symbol = normalize_symbol(symbol)
cache_key = f"history:{symbol}:{days}"
if cache_key in self.cache:
return self.cache[cache_key]
today = date.today()
params = {
"symbol": symbol,
"from": (today - timedelta(days=days * 2 + 10)).isoformat(),
"to": today.isoformat(),
}
data = self._get(f"{self.STABLE_URL}/historical-price-eod/full", params, quiet=True)
bars = normalize_price_bars(data, symbol=symbol)
if bars:
result = [bar_to_dict(b) for b in bars[-days:]]
self.cache[cache_key] = result
return result
data = self._get(f"{self.V3_URL}/historical-price-full/{symbol}", {"timeseries": days})
bars = normalize_price_bars(data, symbol=symbol)
result = [bar_to_dict(b) for b in bars[-days:]]
self.cache[cache_key] = result
return result
def stats(self) -> dict[str, Any]:
return {
"api_calls_made": self.api_calls_made,
"max_api_calls": self.max_api_calls,
"budget_remaining": max(0, self.max_api_calls - self.api_calls_made),
}
def normalize_symbol(value: Any) -> str:
return str(value or "").strip().upper().replace(".", "-")
def to_float(value: Any, default: float = 0.0) -> float:
try:
if value is None or value == "":
return default
result = float(value)
if math.isnan(result) or math.isinf(result):
return default
return result
except (TypeError, ValueError):
return default
def to_int(value: Any, default: int = 0) -> int:
try:
if value is None or value == "":
return default
return int(float(value))
except (TypeError, ValueError):
return default
def bar_to_dict(bar: Bar) -> dict[str, Any]:
return {
"date": bar.date,
"open": bar.open,
"high": bar.high,
"low": bar.low,
"close": bar.close,
"volume": bar.volume,
}
def parse_date(value: str) -> date:
return datetime.strptime(value[:10], "%Y-%m-%d").date()
def normalize_price_bars(payload: Any, symbol: str | None = None) -> list[Bar]:
"""Normalize several common OHLCV JSON shapes into oldest -> newest bars.
Supported shapes:
- [{date, open, high, low, close, volume}, ...]
- {"historical": [...]}
- {"prices": {"AAPL": [...]}} handled by load_prices_json, not here
- FMP stable flat list that may include a per-row symbol field
"""
rows: Any
if isinstance(payload, dict) and "historical" in payload:
rows = payload.get("historical") or []
elif isinstance(payload, dict) and "bars" in payload:
rows = payload.get("bars") or []
elif isinstance(payload, list):
rows = payload
else:
return []
bars = []
target = normalize_symbol(symbol) if symbol else None
for row in rows:
if not isinstance(row, dict):
continue
row_symbol = normalize_symbol(row.get("symbol")) if row.get("symbol") else None
if target and row_symbol and row_symbol != target:
continue
dt = str(row.get("date") or row.get("datetime") or "")[:10]
if len(dt) != 10:
continue
open_ = to_float(row.get("open"), default=to_float(row.get("close")))
high = to_float(row.get("high"), default=max(open_, to_float(row.get("close"))))
low = to_float(row.get("low"), default=min(open_, to_float(row.get("close"))))
close = to_float(row.get("close"), default=to_float(row.get("adjClose")))
volume = to_int(row.get("volume"))
if close <= 0:
continue
# If high/low are absent in a close-only feed, keep them equal to close so
# forward return still works but MFE/MAE is conservative.
high = high if high > 0 else close
low = low if low > 0 else close
bars.append(Bar(date=dt, open=open_, high=high, low=low, close=close, volume=volume))
dedup: dict[str, Bar] = {b.date: b for b in bars}
return [dedup[k] for k in sorted(dedup.keys())]
def load_prices_json(path: str | Path) -> dict[str, list[Bar]]:
"""Load offline prices by symbol for no-API outcome updates."""
with open(path, encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, dict) and "prices" in data and isinstance(data["prices"], dict):
data = data["prices"]
result: dict[str, list[Bar]] = {}
if isinstance(data, dict):
for symbol, payload in data.items():
norm = normalize_symbol(symbol)
bars = normalize_price_bars(payload, symbol=norm)
if bars:
result[norm] = bars
elif isinstance(data, list):
grouped: dict[str, list[dict[str, Any]]] = {}
for row in data:
if not isinstance(row, dict):
continue
symbol = normalize_symbol(row.get("symbol") or row.get("ticker"))
if not symbol:
continue
grouped.setdefault(symbol, []).append(row)
for symbol, rows in grouped.items():
bars = normalize_price_bars(rows, symbol=symbol)
if bars:
result[symbol] = bars
return result
def load_model_book(path: str | Path) -> list[dict[str, Any]]:
path = Path(path)
if not path.exists():
return []
records = []
with open(path, encoding="utf-8") as f:
for line_no, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
record = json.loads(line)
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSONL at {path}:{line_no}: {exc}") from exc
if isinstance(record, dict):
records.append(record)
return records
def save_model_book(path: str | Path, records: list[dict[str, Any]]) -> None:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
records = sorted(
records,
key=lambda r: (r.get("setup_date", ""), r.get("symbol", ""), r.get("record_id", "")),
)
with open(path, "w", encoding="utf-8") as f:
for record in records:
f.write(json.dumps(record, sort_keys=True, ensure_ascii=False) + "\n")
def upsert_records(
existing: list[dict[str, Any]],
new_records: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], dict[str, int]]:
by_id = {r.get("record_id"): dict(r) for r in existing if r.get("record_id")}
inserted = 0
updated = 0
for record in new_records:
record_id = record["record_id"]
if record_id in by_id:
previous = by_id[record_id]
merged = dict(previous)
# Refresh deterministic setup fields while preserving human annotations
# and previously calculated outcomes. Fresh ingest records carry empty
# PENDING outcome placeholders, so they should not erase evidence that
# has already matured.
preserved = {
key: previous[key]
for key in ("human_label", "human_decision", "human_notes")
if key in previous
}
previous_has_outcome = bool(previous.get("outcomes")) or previous.get("matured")
if previous_has_outcome and not record.get("outcomes"):
for key in ("outcomes", "overall_outcome", "matured", "last_outcome_update_at"):
if key in previous:
preserved[key] = previous[key]
merged.update(record)
merged.update(preserved)
by_id[record_id] = merged
updated += 1
else:
by_id[record_id] = record
inserted += 1
return list(by_id.values()), {"inserted": inserted, "updated": updated, "total": len(by_id)}
def load_screener_candidates(
paths: Iterable[str | Path], include_rejects: bool = False
) -> list[dict[str, Any]]:
candidates = []
for path in paths:
with open(path, encoding="utf-8") as f:
data = json.load(f)
report_candidates = data.get("candidates") if isinstance(data, dict) else None
if not isinstance(report_candidates, list):
raise ValueError(f"{path} does not look like a stockbee momentum burst JSON report")
metadata = data.get("metadata", {}) if isinstance(data.get("metadata", {}), dict) else {}
for c in report_candidates:
if not isinstance(c, dict):
continue
if not include_rejects:
state = str(c.get("state", "")).upper()
if state in {"REJECT", "REJECTED", "INVALID"}:
continue
item = dict(c)
item["_source_report"] = str(path)
item["_source_metadata"] = metadata
candidates.append(item)
return candidates
def derive_setup_tags(candidate: dict[str, Any]) -> list[str]:
tags: set[str] = set()
for tag in candidate.get("trigger_tags", []) or []:
if tag:
tags.add(str(tag))
for tag in candidate.get("soft_failure_tags", []) or []:
if tag:
tags.add(f"soft_{tag}")
for tag in candidate.get("reject_reasons", []) or []:
if tag:
tags.add(f"reject_{tag}")
rating = str(candidate.get("rating") or "")
if rating:
tags.add(f"rating_{rating.replace('+', 'plus').replace('-', 'minus')}")
close_location = to_float(candidate.get("close_location_pct"))
if close_location >= 85:
tags.add("close_near_high")
elif close_location < 50:
tags.add("weak_close_location")
volume_ratio = to_float(candidate.get("volume_ratio_1d"))
if volume_ratio >= 3:
tags.add("high_volume_expansion")
elif 0 < volume_ratio < 1.2:
tags.add("thin_volume_confirmation")
base_width = to_float(candidate.get("base_width_pct"))
if 0 < base_width <= 8:
tags.add("tight_base")
elif base_width >= 15:
tags.add("wide_base")
prior_base_days = to_int(candidate.get("prior_base_days"))
if 3 <= prior_base_days <= 20:
tags.add("controlled_3_to_20_day_base")
elif prior_base_days > 30:
tags.add("long_base")
risk = to_float(candidate.get("risk_pct_to_stop"))
if 0 < risk <= 3:
tags.add("compact_risk")
elif risk >= 6:
tags.add("wide_risk")
if candidate.get("volume_dry_up"):
tags.add("volume_dry_up")
if candidate.get("recent_4pct_breakdown"):
tags.add("recent_4pct_breakdown")
if to_int(candidate.get("prior_up_streak")) >= 3:
tags.add("three_days_up_before_trigger")
return sorted(tags)
def make_record(
candidate: dict[str, Any], source_skill: str = "stockbee-momentum-burst-screener"
) -> dict[str, Any]:
symbol = normalize_symbol(candidate.get("symbol"))
setup_date = str(candidate.get("date") or candidate.get("setup_date") or "")[:10]
if not symbol or len(setup_date) != 10:
raise ValueError(f"candidate missing symbol or setup date: {candidate}")
primary_trigger = str(candidate.get("primary_trigger") or "unknown")
record_id = f"stockbee_mb:{symbol}:{setup_date}:{primary_trigger}"
entry = to_float(candidate.get("entry_reference"), default=to_float(candidate.get("close")))
stop = to_float(candidate.get("stop_reference"), default=to_float(candidate.get("low")))
return {
"schema_version": SCHEMA_VERSION,
"record_id": record_id,
"source_skill": source_skill,
"source_report": candidate.get("_source_report"),
"created_at": datetime.now().isoformat(timespec="seconds"),
"updated_at": datetime.now().isoformat(timespec="seconds"),
"symbol": symbol,
"setup_date": setup_date,
"setup_type": "stockbee_momentum_burst",
"primary_trigger": primary_trigger,
"trigger_tags": candidate.get("trigger_tags", []) or [],
"setup_tags": derive_setup_tags(candidate),
"rating": candidate.get("rating"),
"setup_score": to_float(candidate.get("setup_score")),
"state_at_ingest": candidate.get("state"),
"entry_reference": entry,
"stop_reference": stop,
"risk_pct_to_stop": to_float(candidate.get("risk_pct_to_stop")),
"day_gain_pct": to_float(candidate.get("day_gain_pct")),
"volume_ratio_1d": to_float(candidate.get("volume_ratio_1d")),
"volume_ratio_20d": to_float(candidate.get("volume_ratio_20d")),
"close_location_pct": to_float(candidate.get("close_location_pct")),
"prior_base_days": to_int(candidate.get("prior_base_days")),
"base_width_pct": to_float(candidate.get("base_width_pct")),
"human_label": None,
"human_decision": "unknown",
"human_notes": "",
"outcomes": {},
"overall_outcome": "PENDING",
"matured": False,
"raw_candidate": candidate,
}
def is_studyable_candidate(candidate: dict[str, Any]) -> tuple[bool, str | None]:
symbol = normalize_symbol(candidate.get("symbol"))
setup_date = str(candidate.get("date") or candidate.get("setup_date") or "")[:10]
if not symbol:
return False, "missing_symbol"
if len(setup_date) != 10:
return False, "missing_setup_date"
return True, None
def make_studyable_records(
candidates: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], dict[str, int]]:
records = []
stats = {"candidates_loaded": len(candidates), "studyable": 0, "skipped": 0}
for candidate in candidates:
studyable, reason = is_studyable_candidate(candidate)
if not studyable:
stats["skipped"] += 1
key = f"skipped_{reason or 'unknown'}"
stats[key] = stats.get(key, 0) + 1
continue
records.append(make_record(candidate))
stats["studyable"] += 1
return records, stats
def future_bars_after_setup(bars: list[Bar], setup_date: str) -> list[Bar]:
setup_dt = parse_date(setup_date)
return [bar for bar in bars if parse_date(bar.date) > setup_dt]
def calculate_window_outcome(
record: dict[str, Any],
bars: list[Bar],
horizon: int,
) -> dict[str, Any]:
"""Calculate forward close return plus MFE/MAE for N future trading days."""
entry = to_float(record.get("entry_reference"))
stop = to_float(record.get("stop_reference"))
if entry <= 0:
return {"horizon_days": horizon, "matured": False, "reason": "missing_entry"}
future = future_bars_after_setup(bars, str(record.get("setup_date")))
if len(future) < horizon:
return {
"horizon_days": horizon,
"matured": False,
"available_future_bars": len(future),
"reason": "not_enough_future_bars",
}
window = future[:horizon]
close_bar = window[-1]
max_high = max(bar.high for bar in window)
min_low = min(bar.low for bar in window)
fwd = ((close_bar.close / entry) - 1.0) * 100.0
mfe = ((max_high / entry) - 1.0) * 100.0
mae = ((min_low / entry) - 1.0) * 100.0
stop_hit = False
stop_hit_date = None
if stop > 0:
for bar in window:
if bar.low <= stop:
stop_hit = True
stop_hit_date = bar.date
break
tag = classify_outcome(fwd_return_pct=fwd, mfe_pct=mfe, mae_pct=mae, stop_hit=stop_hit)
return {
"horizon_days": horizon,
"matured": True,
"close_date": close_bar.date,
"close": round(close_bar.close, 4),
"forward_return_pct": round(fwd, 2),
"mfe_pct": round(mfe, 2),
"mae_pct": round(mae, 2),
"max_high": round(max_high, 4),
"min_low": round(min_low, 4),
"stop_hit": stop_hit,
"stop_hit_date": stop_hit_date,
"outcome_tag": tag,
}
def classify_outcome(fwd_return_pct: float, mfe_pct: float, mae_pct: float, stop_hit: bool) -> str:
if stop_hit:
return "FAILED_STOP"
if mfe_pct >= 12 or fwd_return_pct >= 8:
return "STRONG_WINNER"
if mfe_pct >= 6 or fwd_return_pct >= 4:
return "WORKED"
if fwd_return_pct <= -2:
return "FAILED_FADE"
if mae_pct <= -5 and fwd_return_pct < 2:
return "CHOPPY_FAILURE"
return "NEUTRAL"
def derive_overall_outcome(outcomes: dict[str, dict[str, Any]], primary_horizon: int = 5) -> str:
preferred = outcomes.get(f"{primary_horizon}d")
if not preferred or not preferred.get("matured"):
return "PENDING"
return str(preferred.get("outcome_tag", "UNKNOWN"))
def update_record_outcomes(
records: list[dict[str, Any]],
prices_by_symbol: dict[str, list[Bar]],
horizons: tuple[int, ...] = DEFAULT_HORIZONS,
only_pending: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
updated = 0
skipped_no_prices = 0
still_pending = 0
matured = 0
for record in records:
if only_pending and record.get("matured"):
continue
symbol = normalize_symbol(record.get("symbol"))
bars = prices_by_symbol.get(symbol)
if not bars:
record["update_error"] = "missing_prices"
skipped_no_prices += 1
continue
outcomes = dict(record.get("outcomes") or {})
for horizon in horizons:
outcomes[f"{horizon}d"] = calculate_window_outcome(record, bars, horizon)
record["outcomes"] = outcomes
record["overall_outcome"] = derive_overall_outcome(outcomes, primary_horizon=max(horizons))
record["matured"] = record["overall_outcome"] != "PENDING"
record["updated_at"] = datetime.now().isoformat(timespec="seconds")
record["last_outcome_update_at"] = record["updated_at"]
if record["matured"]:
matured += 1
else:
still_pending += 1
updated += 1
return records, {
"updated": updated,
"matured": matured,
"still_pending": still_pending,
"skipped_no_prices": skipped_no_prices,
}
def values_for_metric(records: list[dict[str, Any]], metric_path: str) -> list[float]:
parts = metric_path.split(".")
values = []
for record in records:
node: Any = record
for part in parts:
if not isinstance(node, dict):
node = None
break
node = node.get(part)
value = to_float(node, default=float("nan"))
if not math.isnan(value):
values.append(value)
return values
def group_records(records: list[dict[str, Any]], group_by: str) -> dict[str, list[dict[str, Any]]]:
groups: dict[str, list[dict[str, Any]]] = {}
for record in records:
if group_by == "setup_tags":
tags = record.get("setup_tags") or ["untagged"]
for tag in tags:
groups.setdefault(str(tag), []).append(record)
else:
key = record.get(group_by) or "unknown"
groups.setdefault(str(key), []).append(record)
return groups
def summarize_group(records: list[dict[str, Any]]) -> dict[str, Any]:
matured = [r for r in records if r.get("matured")]
sample = len(matured)
if sample == 0:
return {"sample_size": 0, "pending": len(records), "win_rate_pct": None}
winners = [r for r in matured if r.get("overall_outcome") in {"WORKED", "STRONG_WINNER"}]
failures = [
r
for r in matured
if str(r.get("overall_outcome", "")).startswith("FAILED")
or r.get("overall_outcome") == "CHOPPY_FAILURE"
]
strong = [r for r in matured if r.get("overall_outcome") == "STRONG_WINNER"]
stop = [r for r in matured if r.get("outcomes", {}).get("5d", {}).get("stop_hit")]
ret3 = values_for_metric(matured, "outcomes.3d.forward_return_pct")
ret5 = values_for_metric(matured, "outcomes.5d.forward_return_pct")
mfe5 = values_for_metric(matured, "outcomes.5d.mfe_pct")
mae5 = values_for_metric(matured, "outcomes.5d.mae_pct")
return {
"sample_size": sample,
"pending": len(records) - sample,
"win_rate_pct": round((len(winners) / sample) * 100, 1),
"failure_rate_pct": round((len(failures) / sample) * 100, 1),
"strong_winner_rate_pct": round((len(strong) / sample) * 100, 1),
"stop_hit_rate_pct": round((len(stop) / sample) * 100, 1),
"avg_3d_return_pct": round(statistics.fmean(ret3), 2) if ret3 else None,
"avg_5d_return_pct": round(statistics.fmean(ret5), 2) if ret5 else None,
"median_5d_return_pct": round(statistics.median(ret5), 2) if ret5 else None,
"avg_5d_mfe_pct": round(statistics.fmean(mfe5), 2) if mfe5 else None,
"avg_5d_mae_pct": round(statistics.fmean(mae5), 2) if mae5 else None,
}
def summarize_model_book(
records: list[dict[str, Any]],
group_by_fields: list[str],
min_sample: int = 3,
) -> dict[str, Any]:
matured = [r for r in records if r.get("matured")]
summary: dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"skill": SKILL_NAME,
"generated_at": datetime.now().isoformat(timespec="seconds"),
"total_records": len(records),
"matured_records": len(matured),
"pending_records": len(records) - len(matured),
"overall": summarize_group(records),
"groups": {},
"rule_candidates": [],
}
for group_by in group_by_fields:
groups = group_records(records, group_by)
rows = []
for key, group in groups.items():
stats = summarize_group(group)
if stats.get("sample_size", 0) < min_sample:
continue
row = {"group": key, **stats}
rows.append(row)
if is_promote_candidate(row):
summary["rule_candidates"].append(
{
"action": "promote",
"group_by": group_by,
"group": key,
"reason": "High win rate with positive 5d expectancy and acceptable drawdown.",
"evidence": row,
}
)
elif is_downgrade_candidate(row):
summary["rule_candidates"].append(
{
"action": "downgrade_or_filter",
"group_by": group_by,
"group": key,
"reason": "Weak 5d expectancy or high failure rate. Review examples before changing rules.",
"evidence": row,
}
)
rows.sort(
key=lambda r: (
r.get("avg_5d_return_pct") if r.get("avg_5d_return_pct") is not None else -999,
r.get("win_rate_pct") if r.get("win_rate_pct") is not None else -999,
),
reverse=True,
)
summary["groups"][group_by] = rows
return summary
def is_promote_candidate(row: dict[str, Any]) -> bool:
sample = row.get("sample_size") or 0
win_rate = row.get("win_rate_pct") or 0
avg5 = row.get("avg_5d_return_pct") or 0
mae = row.get("avg_5d_mae_pct") or 0
return sample >= 5 and win_rate >= 60 and avg5 >= 2 and mae > -5
def is_downgrade_candidate(row: dict[str, Any]) -> bool:
sample = row.get("sample_size") or 0
fail_rate = row.get("failure_rate_pct") or 0
avg5 = row.get("avg_5d_return_pct") or 0
return sample >= 5 and (fail_rate >= 50 or avg5 <= -1)
def write_json_report(payload: dict[str, Any], output_dir: str | Path, prefix: str) -> Path:
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now().strftime("%Y-%m-%d_%H%M%S")
path = output_dir / f"{prefix}_{ts}.json"
with open(path, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False)
return path
def write_markdown_report(payload: dict[str, Any], output_dir: str | Path, prefix: str) -> Path:
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now().strftime("%Y-%m-%d_%H%M%S")
path = output_dir / f"{prefix}_{ts}.md"
lines = []
title = payload.get("title") or "Stockbee Setup Fluency Report"
lines.append(f"# {title}")
lines.append("")
lines.append(
f"Generated: {payload.get('generated_at') or datetime.now().isoformat(timespec='seconds')}"
)
lines.append("")
if "ingest" in payload:
stats = payload["ingest"]
lines.append("## Ingest")
lines.append("")
lines.append(f"- Candidates loaded: {stats.get('candidates_loaded', 0)}")
lines.append(f"- Studyable records: {stats.get('studyable', 0)}")
lines.append(f"- Skipped: {stats.get('skipped', 0)}")
if stats.get("skipped_missing_setup_date"):
lines.append(
f"- Skipped missing setup date: {stats.get('skipped_missing_setup_date', 0)}"
)
if stats.get("skipped_missing_symbol"):
lines.append(f"- Skipped missing symbol: {stats.get('skipped_missing_symbol', 0)}")
lines.append(f"- Inserted: {stats.get('inserted', 0)}")
lines.append(f"- Updated: {stats.get('updated', 0)}")
lines.append(f"- Total model-book records: {stats.get('total', 0)}")
lines.append("")
if "update" in payload:
stats = payload["update"]
lines.append("## Outcome Update")
lines.append("")
lines.append(f"- Updated: {stats.get('updated', 0)}")
lines.append(f"- Matured: {stats.get('matured', 0)}")
lines.append(f"- Still pending: {stats.get('still_pending', 0)}")
lines.append(f"- Missing prices: {stats.get('skipped_no_prices', 0)}")
lines.append("")
if "overall" in payload:
overall = payload.get("overall") or {}
lines.append("## Overall")
lines.append("")
lines.append(f"- Total records: {payload.get('total_records', 0)}")
lines.append(f"- Matured records: {payload.get('matured_records', 0)}")
lines.append(f"- Pending records: {payload.get('pending_records', 0)}")
if overall.get("sample_size"):
lines.append(f"- Win rate: {overall.get('win_rate_pct')}%")
lines.append(f"- Avg 5d return: {overall.get('avg_5d_return_pct')}%")
lines.append(
f"- Avg 5d MFE / MAE: {overall.get('avg_5d_mfe_pct')}% / {overall.get('avg_5d_mae_pct')}%"
)
lines.append("")
groups = payload.get("groups") or {}
for group_by, rows in groups.items():
lines.append(f"## Cohorts by {group_by}")
lines.append("")
if not rows:
lines.append("No cohorts met the sample threshold.")
lines.append("")
continue
lines.append("| Group | N | Win% | Fail% | Avg 5d | Avg MFE | Avg MAE |")
lines.append("|---|---:|---:|---:|---:|---:|---:|")
for row in rows[:20]:
lines.append(
f"| {row.get('group')} | {row.get('sample_size')} | {row.get('win_rate_pct')} | "
f"{row.get('failure_rate_pct')} | {row.get('avg_5d_return_pct')} | "
f"{row.get('avg_5d_mfe_pct')} | {row.get('avg_5d_mae_pct')} |"
)
lines.append("")
rules = payload.get("rule_candidates") or []
if rules:
lines.append("## Rule Candidates")
lines.append("")
for rule in rules[:20]:
evidence = rule.get("evidence", {})
lines.append(
f"- **{rule.get('action')}** `{rule.get('group_by')}={rule.get('group')}`: "
f"{rule.get('reason')} N={evidence.get('sample_size')}, "
f"Win={evidence.get('win_rate_pct')}%, Avg5d={evidence.get('avg_5d_return_pct')}%."
)
lines.append("")
lines.append("## Reminder")
lines.append("")
lines.append(
"This model book is a learning dataset, not an execution engine. Promote or downgrade rules only after manual chart review of representative examples."
)
lines.append("")
with open(path, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
return path
def parse_horizons(value: str) -> tuple[int, ...]:
horizons = []
for part in value.split(","):
part = part.strip()
if not part:
continue
n = int(part)
if n <= 0:
raise ValueError("horizons must be positive integers")
horizons.append(n)
if not horizons:
raise ValueError("at least one horizon is required")
return tuple(sorted(set(horizons)))
def command_ingest(args: argparse.Namespace) -> int:
candidates = load_screener_candidates(args.screener_json, include_rejects=args.include_rejects)
new_records, ingest_stats = make_studyable_records(candidates)
existing = load_model_book(args.model_book)
records, upsert_stats = upsert_records(existing, new_records)
stats = {**ingest_stats, **upsert_stats}
save_model_book(args.model_book, records)
payload = {
"schema_version": SCHEMA_VERSION,
"skill": SKILL_NAME,
"title": "Stockbee Setup Fluency Ingest Report",
"generated_at": datetime.now().isoformat(timespec="seconds"),
"model_book": args.model_book,
"source_reports": args.screener_json,
"candidates_loaded": len(candidates),
"ingest": stats,
}
json_path = write_json_report(payload, args.output_dir, "stockbee_setup_fluency_ingest")
md_path = write_markdown_report(payload, args.output_dir, "stockbee_setup_fluency_ingest")
print(
f"Ingested {stats.get('studyable', 0)} studyable candidates into {args.model_book} "
f"({stats.get('skipped', 0)} skipped)"
)
print(f"JSON Report: {json_path}")
print(f"Markdown Report: {md_path}")
return 0
def build_prices_for_update(
args: argparse.Namespace, records: list[dict[str, Any]]
) -> tuple[dict[str, list[Bar]], dict[str, Any] | None]:
if args.prices_json:
return load_prices_json(args.prices_json), None
client = FMPClient(api_key=args.api_key, max_api_calls=args.max_api_calls)
prices: dict[str, list[Bar]] = {}
symbols = sorted(
{normalize_symbol(r.get("symbol")) for r in records if normalize_symbol(r.get("symbol"))}
)
for symbol in symbols:
raw = client.get_historical_prices(symbol, days=args.history_days)
bars = normalize_price_bars(raw, symbol=symbol)
if bars:
prices[symbol] = bars
return prices, client.stats()
def command_update(args: argparse.Namespace) -> int:
records = load_model_book(args.model_book)
if not records:
print(f"No records found in {args.model_book}")
return 0
prices, api_stats = build_prices_for_update(args, records)
horizons = parse_horizons(args.horizons)
records, stats = update_record_outcomes(
records, prices, horizons=horizons, only_pending=args.only_pending
)
save_model_book(args.model_book, records)
payload = {
"schema_version": SCHEMA_VERSION,
"skill": SKILL_NAME,
"title": "Stockbee Setup Fluency Outcome Update",
"generated_at": datetime.now().isoformat(timespec="seconds"),
"model_book": args.model_book,
"horizons": list(horizons),
"prices_source": args.prices_json or "fmp",
"update": stats,
"api_stats": api_stats,
}
json_path = write_json_report(payload, args.output_dir, "stockbee_setup_fluency_update")
md_path = write_markdown_report(payload, args.output_dir, "stockbee_setup_fluency_update")
print(f"Updated outcomes in {args.model_book}")
print(f"JSON Report: {json_path}")
print(f"Markdown Report: {md_path}")
return 0
def command_summarize(args: argparse.Namespace) -> int:
records = load_model_book(args.model_book)
group_by = [part.strip() for part in args.group_by.split(",") if part.strip()]
summary = summarize_model_book(records, group_by_fields=group_by, min_sample=args.min_sample)
summary["model_book"] = args.model_book
json_path = write_json_report(summary, args.output_dir, "stockbee_setup_fluency_summary")
md_path = write_markdown_report(summary, args.output_dir, "stockbee_setup_fluency_summary")
print(f"Summarized {len(records)} records from {args.model_book}")
print(f"JSON Report: {json_path}")
print(f"Markdown Report: {md_path}")
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Stockbee setup fluency model-book trainer")
sub = parser.add_subparsers(dest="command", required=True)
ingest = sub.add_parser("ingest", help="Ingest Stockbee momentum-burst screener reports")
ingest.add_argument(
"--screener-json", nargs="+", required=True, help="Momentum burst JSON report(s)"
)
ingest.add_argument("--model-book", default=DEFAULT_MODEL_BOOK, help="JSONL model-book path")
ingest.add_argument("--output-dir", default="reports/", help="Report output directory")
ingest.add_argument(
"--include-rejects",
action="store_true",
help="Also ingest rejected candidates as negative examples",
)
ingest.set_defaults(func=command_ingest)
update = sub.add_parser(
"update", help="Update pending model-book records with forward outcomes"
)
update.add_argument("--model-book", default=DEFAULT_MODEL_BOOK, help="JSONL model-book path")
update.add_argument("--prices-json", help="Offline OHLCV JSON by symbol; avoids FMP")
update.add_argument("--api-key", help="FMP API key; defaults to FMP_API_KEY")
update.add_argument("--max-api-calls", type=int, default=200, help="FMP call budget")
update.add_argument(
"--history-days", type=int, default=80, help="Daily bars per symbol when using FMP"
)
update.add_argument(
"--horizons", default="3,5", help="Comma-separated forward horizons in trading days"
)
update.add_argument("--output-dir", default="reports/", help="Report output directory")
update.add_argument(
"--only-pending",
action=argparse.BooleanOptionalAction,
default=True,
help="Only update records not yet matured",
)
update.set_defaults(func=command_update)
summarize = sub.add_parser("summarize", help="Aggregate model-book evidence into cohort stats")
summarize.add_argument("--model-book", default=DEFAULT_MODEL_BOOK, help="JSONL model-book path")
summarize.add_argument(
"--group-by",
default="rating,primary_trigger,setup_tags",
help="Comma-separated fields or setup_tags",
)
summarize.add_argument(
"--min-sample", type=int, default=3, help="Minimum matured records per cohort"
)
summarize.add_argument("--output-dir", default="reports/", help="Report output directory")
summarize.set_defaults(func=command_summarize)
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
return args.func(args)
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())
scripts/tests/test_build_model_book.py
import json
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(SCRIPT_DIR))
from build_model_book import ( # noqa: E402
Bar,
classify_outcome,
load_model_book,
load_prices_json,
load_screener_candidates,
make_record,
make_studyable_records,
save_model_book,
summarize_model_book,
update_record_outcomes,
upsert_records,
)
def candidate(symbol="TEST", date="2026-06-20", rating="A"):
return {
"symbol": symbol,
"date": date,
"state": "ACTIONABLE_DAY1",
"rating": rating,
"setup_score": 93,
"primary_trigger": "4pct_breakout",
"trigger_tags": ["4pct_breakout", "dollar_breakout", "range_expansion"],
"entry_reference": 104.5,
"stop_reference": 101.0,
"risk_pct_to_stop": 3.35,
"day_gain_pct": 4.5,
"volume_ratio_1d": 5.0,
"volume_ratio_20d": 3.63,
"close_location_pct": 87.5,
"prior_base_days": 8,
"base_width_pct": 3.2,
"volume_dry_up": True,
"recent_4pct_breakdown": False,
"prior_up_streak": 1,
}
def test_make_record_derives_model_book_fields_and_tags():
record = make_record(candidate())
assert record["record_id"] == "stockbee_mb:TEST:2026-06-20:4pct_breakout"
assert record["overall_outcome"] == "PENDING"
assert record["matured"] is False
assert "close_near_high" in record["setup_tags"]
assert "high_volume_expansion" in record["setup_tags"]
assert "tight_base" in record["setup_tags"]
assert "controlled_3_to_20_day_base" in record["setup_tags"]
assert "volume_dry_up" in record["setup_tags"]
def test_include_rejects_skips_skeleton_rejects_without_date(tmp_path):
path = tmp_path / "screener.json"
path.write_text(
json.dumps(
{
"metadata": {"source": "test"},
"candidates": [
candidate(),
{
"symbol": "SKEL",
"state": "REJECT",
"reject_reasons": ["insufficient_history"],
},
],
}
),
encoding="utf-8",
)
candidates = load_screener_candidates([path], include_rejects=True)
records, stats = make_studyable_records(candidates)
assert len(candidates) == 2
assert len(records) == 1
assert stats["studyable"] == 1
assert stats["skipped"] == 1
assert stats["skipped_missing_setup_date"] == 1
def test_update_outcomes_calculates_forward_return_mfe_mae_and_strong_winner():
record = make_record(candidate())
bars = [
Bar("2026-06-20", 100, 105, 101, 104.5, 600000),
Bar("2026-06-23", 105, 109, 104, 108, 700000),
Bar("2026-06-24", 108, 112, 107, 111, 800000),
Bar("2026-06-25", 111, 118, 110, 116, 900000),
Bar("2026-06-26", 116, 119, 114, 118, 850000),
Bar("2026-06-29", 118, 121, 117, 120, 900000),
]
records, stats = update_record_outcomes([record], {"TEST": bars}, horizons=(3, 5))
updated = records[0]
assert stats["updated"] == 1
assert stats["matured"] == 1
assert updated["matured"] is True
assert updated["overall_outcome"] == "STRONG_WINNER"
assert updated["outcomes"]["3d"]["forward_return_pct"] == 11.0
assert updated["outcomes"]["5d"]["mfe_pct"] == 15.79
assert updated["outcomes"]["5d"]["mae_pct"] == -0.48
def test_update_outcomes_marks_stop_failure():
record = make_record(candidate())
bars = [
Bar("2026-06-20", 100, 105, 101, 104.5, 600000),
Bar("2026-06-23", 104, 105, 100.5, 101.5, 700000),
Bar("2026-06-24", 101, 102, 98, 99, 800000),
Bar("2026-06-25", 99, 100, 96, 97, 900000),
Bar("2026-06-26", 97, 98, 95, 96, 850000),
Bar("2026-06-29", 96, 97, 94, 95, 900000),
]
records, _ = update_record_outcomes([record], {"TEST": bars}, horizons=(3, 5))
updated = records[0]
assert updated["outcomes"]["3d"]["stop_hit"] is True
assert updated["outcomes"]["3d"]["stop_hit_date"] == "2026-06-23"
assert updated["overall_outcome"] == "FAILED_STOP"
def test_classify_outcome_handles_neutral_and_worked():
assert classify_outcome(4.2, 6.1, -1.0, stop_hit=False) == "WORKED"
assert classify_outcome(0.5, 2.0, -1.0, stop_hit=False) == "NEUTRAL"
assert classify_outcome(-2.5, 1.0, -4.0, stop_hit=False) == "FAILED_FADE"
def test_upsert_records_preserves_existing_outcome_fields():
record = make_record(candidate())
record["outcomes"] = {"5d": {"matured": True, "outcome_tag": "WORKED"}}
record["overall_outcome"] = "WORKED"
record["matured"] = True
fresh = make_record(candidate())
fresh["setup_score"] = 95
records, stats = upsert_records([record], [fresh])
assert stats == {"inserted": 0, "updated": 1, "total": 1}
assert records[0]["setup_score"] == 95
assert records[0]["overall_outcome"] == "WORKED"
assert records[0]["matured"] is True
def test_load_prices_json_supports_prices_mapping(tmp_path):
path = tmp_path / "prices.json"
payload = {
"prices": {
"TEST": [
{"date": "2026-06-24", "open": 108, "high": 112, "low": 107, "close": 111},
{"date": "2026-06-23", "open": 105, "high": 109, "low": 104, "close": 108},
]
}
}
path.write_text(json.dumps(payload), encoding="utf-8")
prices = load_prices_json(path)
assert list(prices) == ["TEST"]
assert [b.date for b in prices["TEST"]] == ["2026-06-23", "2026-06-24"]
def test_save_and_load_model_book_jsonl(tmp_path):
path = tmp_path / "model_book.jsonl"
record = make_record(candidate())
save_model_book(path, [record])
loaded = load_model_book(path)
assert loaded[0]["record_id"] == record["record_id"]
def test_summarize_model_book_groups_by_tags():
records = []
for i in range(6):
r = make_record(candidate(symbol=f"T{i}", rating="A"))
r["matured"] = True
r["overall_outcome"] = "WORKED" if i < 5 else "NEUTRAL"
r["outcomes"] = {
"3d": {"forward_return_pct": 3.0, "matured": True},
"5d": {
"forward_return_pct": 4.0 if i < 5 else 0.5,
"mfe_pct": 7.0,
"mae_pct": -2.0,
"stop_hit": False,
"matured": True,
},
}
records.append(r)
summary = summarize_model_book(records, group_by_fields=["rating", "setup_tags"], min_sample=3)
assert summary["matured_records"] == 6
assert summary["groups"]["rating"][0]["group"] == "A"
assert summary["groups"]["rating"][0]["win_rate_pct"] == 83.3
assert any(rule["action"] == "promote" for rule in summary["rule_candidates"])