references/model_book_schema.md
# Stockbee Setup Model Book Schema
The model book is a JSONL file. Each line is one study record for one setup candidate on one setup date.
Default path:
```text
state/stockbee/model_book.jsonl
```
## Record Identity
```json
{
"schema_version": "1.0",
"record_id": "stockbee_mb:TEST:2026-06-20:4pct_breakout",
"source_skill": "stockbee-momentum-burst-screener",
"source_report": "reports/stockbee_momentum_burst_2026-06-20_174954.json",
"symbol": "TEST",
"setup_date": "2026-06-20",
"setup_type": "stockbee_momentum_burst",
"primary_trigger": "4pct_breakout"
}
```
`record_id` is deterministic so repeated ingest runs update the same record instead of duplicating it.
## Setup Quality Fields
```json
{
"rating": "A-",
"setup_score": 84,
"state_at_ingest": "ACTIONABLE_DAY1",
"trigger_tags": ["4pct_breakout", "range_expansion"],
"setup_tags": ["close_near_high", "tight_base", "compact_risk"],
"entry_reference": 52.4,
"stop_reference": 50.9,
"risk_pct_to_stop": 2.86,
"day_gain_pct": 5.22,
"volume_ratio_1d": 2.99,
"close_location_pct": 86.0,
"prior_base_days": 8,
"base_width_pct": 4.8
}
```
The `setup_tags` field is the core of the fluency loop. It turns subjective chart-review features into analyzable cohorts.
## Human Review Fields
```json
{
"human_label": "A-",
"human_decision": "entered",
"human_notes": "Clean 8-day base, high close, theme support."
}
```
The script initializes these fields but does not overwrite human edits during re-ingest.
## Outcome Fields
```json
{
"outcomes": {
"3d": {
"matured": true,
"close_date": "2026-06-25",
"forward_return_pct": 6.42,
"mfe_pct": 8.15,
"mae_pct": -1.20,
"stop_hit": false,
"outcome_tag": "WORKED"
},
"5d": {
"matured": true,
"close_date": "2026-06-29",
"forward_return_pct": 9.11,
"mfe_pct": 12.40,
"mae_pct": -1.20,
"stop_hit": false,
"outcome_tag": "STRONG_WINNER"
}
},
"overall_outcome": "STRONG_WINNER",
"matured": true
}
```
The primary summary outcome uses the longest requested horizon, usually 5 trading days.
## Lifecycle
```text
PENDING_OUTCOME
Ingested, but not enough future bars exist yet.
MATURED_OUTCOME
3-day and/or 5-day windows have enough data.
REVIEWED_BY_HUMAN
Trader has inspected the chart and added human_label / human_notes.
```
The current script uses `matured: true/false`; human review status can be inferred from `human_label` and `human_notes`.
references/outcome_tags.md
# Outcome Tags
The trainer classifies each 3-day and 5-day window into a compact outcome tag. These are learning labels, not trading signals.
## STRONG_WINNER
Conditions:
```text
MFE >= 12% OR forward close return >= 8%
AND stop was not hit
```
Interpretation: The setup produced the kind of fast follow-through Stockbee Momentum Burst is designed to find.
## WORKED
Conditions:
```text
MFE >= 6% OR forward close return >= 4%
AND stop was not hit
```
Interpretation: The setup had tradable follow-through, even if it was not exceptional.
## FAILED_STOP
Condition:
```text
Any low in the horizon <= stop_reference
```
Interpretation: The setup would have violated the planned invalidation level. This is the most important failure type for risk discipline.
## FAILED_FADE
Condition:
```text
forward close return <= -2%
AND stop was not recorded as hit
```
Interpretation: The stock faded after triggering. Review for weak close location, wide base, low volume quality, broad-market weakness, or late-stage extension.
## CHOPPY_FAILURE
Condition:
```text
MAE <= -5%
AND forward close return < 2%
AND stop was not recorded as hit
```
Interpretation: The setup may have looked strong but created too much adverse excursion for a short-term swing process.
## NEUTRAL
Condition:
```text
No decisive follow-through or failure
```
Interpretation: These examples are still useful. A neutral cluster can reveal setups that consume attention without enough reward.
## PENDING
Condition:
```text
Not enough future trading bars are available
```
Interpretation: Leave the record in the model book and update it after the 3-day or 5-day window matures.
references/review_workflow.md
# Stockbee Setup Fluency Review Workflow
This workflow separates candidate generation from learning. The goal is to build procedural recognition of A-quality setups by repeatedly comparing what the setup looked like on day 1 with what happened over the next 3 to 5 trading days.
## Daily After-Close Routine
1. Run `stockbee-momentum-burst-screener`.
2. Ingest the JSON report into the model book.
3. Add human notes for any A/A- setups reviewed manually.
4. Do not change trading rules from one day of examples.
```bash
python3 skills/stockbee-setup-fluency-trainer/scripts/build_model_book.py ingest \
--screener-json reports/stockbee_momentum_burst_*.json \
--model-book state/stockbee/model_book.jsonl \
--output-dir reports/
```
## 3-Day / 5-Day Outcome Routine
Run the update command daily or a few times per week. Matured records will receive forward returns, MFE, MAE, and outcome tags.
```bash
python3 skills/stockbee-setup-fluency-trainer/scripts/build_model_book.py update \
--model-book state/stockbee/model_book.jsonl \
--horizons 3,5 \
--output-dir reports/
```
## Weekly Review
Run a summary with a small sample threshold to see early patterns.
```bash
python3 skills/stockbee-setup-fluency-trainer/scripts/build_model_book.py summarize \
--model-book state/stockbee/model_book.jsonl \
--min-sample 3 \
--output-dir reports/
```
Review:
- Best A/A- winners
- Worst failed-stop examples
- False positives with weak close quality
- Any `wide_risk`, `wide_base`, or `three_days_up_before_trigger` clusters
## Monthly Review
Use a higher sample threshold before changing rules.
```bash
python3 skills/stockbee-setup-fluency-trainer/scripts/build_model_book.py summarize \
--model-book state/stockbee/model_book.jsonl \
--min-sample 10 \
--output-dir reports/
```
Possible actions:
- Promote tags that repeatedly produce positive 5-day expectancy
- Downgrade tags with high stop-hit or fade failure rates
- Add manual chart examples to a separate visual model book
- Feed accepted lessons into `trader-memory-core` or the monthly review process
## Review Discipline
Do not overfit to a tiny sample. Treat `rule_candidates` as prompts for chart review. A rule change should normally require enough examples, plausible market logic, and manual inspection of representative charts.
requirements.txt
# Runtime dependencies for standalone .skill installs (issue #330).
# Verified by scripts/check_skill_deps.py.
requests>=2.31.0
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 importlib.util
import json
import math
import os
import statistics
import sys
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import date, datetime
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."""
_SHARED_FMP_MODULE = "_stockbee_setup_fluency_trainer_fmp_client"
def _load_shared_fmp():
"""Load the sibling generated client by path (never `import fmp_client`)."""
module = sys.modules.get(_SHARED_FMP_MODULE)
if module is None:
path = Path(__file__).resolve().parent / "fmp_client.py"
spec = importlib.util.spec_from_file_location(_SHARED_FMP_MODULE, path)
module = importlib.util.module_from_spec(spec)
sys.modules[_SHARED_FMP_MODULE] = module
try:
spec.loader.exec_module(module)
except BaseException:
sys.modules.pop(_SHARED_FMP_MODULE, None)
raise
return module
def _delegated(name: str) -> property:
def getter(self):
return getattr(self._client, name)
def setter(self, value):
setattr(self._client, name, value)
return property(getter, setter)
class FMPClient:
"""Thin adapter over the generated stable-only FMP client (#495 Batch C)."""
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.")
shared = _load_shared_fmp()
self._budget_error = shared.ApiCallBudgetExceeded
self._client = shared.FMPClient(api_key=self.api_key, max_api_calls=max_api_calls)
self.cache: dict[str, Any] = {}
self.failed_requests = 0
session = _delegated("session")
api_calls_made = _delegated("api_calls_made")
max_api_calls = _delegated("max_api_calls")
last_call_time = _delegated("last_call_time")
rate_limit_reached = _delegated("rate_limit_reached")
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]
try:
data = self._client.fetch_eod_rows(symbol, days, pad_days=10, quiet=False)
except self._budget_error as exc:
raise ApiCallBudgetExceeded(str(exc)) from exc
if data is None:
self.failed_requests += 1
return []
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),
"rate_limit_reached": self.rate_limit_reached,
"failed_requests": self.failed_requests,
}
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 requested_symbols(records: list[dict[str, Any]]) -> list[str]:
return sorted(
{normalize_symbol(r.get("symbol")) for r in records if normalize_symbol(r.get("symbol"))}
)
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 = requested_symbols(records)
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)
if api_stats is not None:
symbol_count = len(requested_symbols(records))
if symbol_count > 0 and api_stats["failed_requests"] == symbol_count:
print(
f"ERROR: FMP price fetch failed for all {symbol_count} symbols; "
"model book not updated",
file=sys.stderr,
)
return 1
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/fmp_client.py
#!/usr/bin/env python3
# GENERATED by scripts/generate_fmp_client.py — do not edit.
# Source of truth: scripts/fmp_client/ (core_template.py.tmpl, registry.py, extensions/).
# Regenerate: python3 scripts/generate_fmp_client.py
"""
FMP API Client for stockbee-setup-fluency-trainer
Provides rate-limited access to Financial Modeling Prep API endpoints.
Features:
- Rate limiting (0.3s between requests)
- Automatic retry on 429 errors
- Session caching for duplicate requests
- Stable-only screener/EOD/quote primitives for stockbee adapters
"""
import os
import re
import sys
import time
from datetime import datetime, timedelta
from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
try:
import requests
except ImportError:
print("ERROR: requests library not found. Install with: pip install requests", file=sys.stderr)
sys.exit(1)
# Mirror of scripts/provider_contracts.py::redact_url (clients are standalone).
# Requires a `?`/`&` prefix, so this masks apikey=/api_key= in URL query
# strings, not a rendered params dict (unreachable today: only
# RequestException text -- which can embed the full request URL -- is
# ever printed here, never a bare params repr()).
_APIKEY_RE = re.compile(r"([?&](?:apikey|api_key)=)[^&\s]+", re.IGNORECASE)
def _redact_key(text: str) -> str:
"""Mask apikey=/api_key= query values so stderr never carries the key."""
return _APIKEY_RE.sub(r"\1REDACTED", str(text))
# --- FMP endpoint fallback: stable (new users) -> v3 (legacy users) ---
def _et_today(now=None):
"""America/New_York calendar date for provider query windows (issue #427).
An explicit aware ``now`` wins (deterministic runs/tests); otherwise the
current instant is converted to America/New_York. ``now`` must be
timezone-aware when supplied. Fail-closed: missing TZ data raises
RuntimeError naming the tzdata requirement instead of silently falling
back to the runner's local clock.
"""
try:
tz = ZoneInfo("America/New_York")
except ZoneInfoNotFoundError as exc:
raise RuntimeError("America/New_York timezone data is missing; pip install tzdata") from exc
if now is not None:
if now.tzinfo is None:
raise ValueError("now must be timezone-aware")
return now.astimezone(tz).date()
return datetime.now(tz).date()
def _stable_hist_url(base, symbols_str, params):
"""stable/historical-price-eod/full?symbol=^GSPC&from=...&to=..."""
params["symbol"] = symbols_str
# New stable EOD endpoint ignores `timeseries`; convert to from/to range
# to bound the payload. Use 2x calendar days to cover N trading days
# (trading-day/calendar-day ratio ~252/365 ~0.69, so *2 leaves headroom).
days = params.pop("timeseries", None)
if days is not None:
today = _et_today()
params["from"] = (today - timedelta(days=int(days) * 2)).isoformat()
params["to"] = today.isoformat()
return base, params
def _v3_hist_url(base, symbols_str, params):
"""api/v3/historical-price-full/^GSPC?timeseries=80"""
return f"{base}/{symbols_str}", params
_FMP_ENDPOINTS = {
"historical": [
("https://financialmodelingprep.com/stable/historical-price-eod/full", _stable_hist_url),
("https://financialmodelingprep.com/api/v3/historical-price-full", _v3_hist_url),
],
}
def _normalize_eod_flat_list(data, symbols_str: str, limit: Optional[int] = None):
"""Convert stable/historical-price-eod/full flat list to v3-compatible dict.
Input : [{"symbol": "SPY", "date": "...", "open": ..., ...}, ...]
Output : {"symbol": "SPY", "historical": [{"date": ..., "open": ..., ...}, ...]}
Returns the input unchanged if not a list (passthrough for v3 dict /
historicalStockList responses). Returns None when no row matches the
requested symbol; the caller will record the failure and try the next
endpoint.
If `limit` is provided (the original `timeseries=N` request), the
`historical` list is truncated to the first `limit` entries. The new
EOD endpoint ignores `timeseries` and returns the full available history,
so the caller's date-range bounding plus this truncation together preserve
the legacy "most-recent N rows" contract. Truncation assumes descending
date order, which the FMP EOD endpoint provides (verified live).
Note: empty list ``[]`` does not reach this normalizer because the caller's
``if not data: continue`` falsy check handles it earlier in
``_request_with_fallback``.
"""
if not isinstance(data, list):
return data
if not data:
return None
norm_target = symbols_str.replace("-", ".")
matched_symbol = None
historical = []
for row in data:
if not isinstance(row, dict):
continue
# Be permissive: single-symbol endpoint may omit per-row "symbol".
# Treat missing symbol as belonging to the requested symbols_str.
row_sym = row.get("symbol") or symbols_str
if row_sym.replace("-", ".") != norm_target:
continue
matched_symbol = matched_symbol or row_sym
historical.append({k: v for k, v in row.items() if k != "symbol"})
if not historical:
return None
if limit is not None and limit > 0:
historical = historical[:limit]
return {"symbol": matched_symbol or symbols_str, "historical": historical}
class ApiCallBudgetExceeded(Exception):
"""Raised when the API call budget has been exhausted."""
pass
class FMPClient:
"""Client for Financial Modeling Prep API with rate limiting, caching, and budget control"""
BASE_URL = "https://financialmodelingprep.com/api/v3"
RATE_LIMIT_DELAY = 0.3 # 300ms between requests
_ENDPOINT_FAILURE_THRESHOLD = 3 # disable endpoint after N consecutive failures
STABLE_URL = "https://financialmodelingprep.com/stable"
def __init__(self, api_key: Optional[str] = None, max_api_calls: int = 200):
self.api_key = (api_key or os.getenv("FMP_API_KEY") or "").strip()
if not self.api_key:
raise ValueError(
"FMP API key required. Set FMP_API_KEY environment variable "
"or pass api_key parameter."
)
self.session = requests.Session()
self.session.headers.update({"apikey": self.api_key})
self.cache = {}
self.last_call_time = 0
self.rate_limit_reached = False
self.retry_count = 0
self.max_retries = 1
self.api_calls_made = 0
self.max_api_calls = max_api_calls
# Circuit breaker: track consecutive failures per endpoint URL prefix
self._endpoint_failures: dict[str, int] = {}
self._disabled_endpoints: set[str] = set()
# Most recent transport-level failure reason; set by _rate_limited_get
# so _request_with_fallback can surface suppressed errors even when
# an endpoint was called with quiet=True.
self._last_error: Optional[str] = None
def _redact_secret(self, text) -> str:
"""Mask raw and URL-encoded key echoes before diagnostics are truncated."""
from urllib.parse import quote, quote_plus
text = _redact_key(text)
if not self.api_key:
return text
patterns = {re.escape(self.api_key)}
for encoded in (quote(self.api_key, safe=""), quote_plus(self.api_key, safe="")):
# Percent triplets are case-insensitive; the key's letters are not.
parts = re.split(r"(%[0-9A-Fa-f]{2})", encoded)
patterns.add(
"".join(
"(?i:" + re.escape(part) + ")"
if re.fullmatch(r"%[0-9A-Fa-f]{2}", part)
else re.escape(part)
for part in parts
)
)
# One pass prevents a short key from matching the replacement marker.
return re.sub("|".join(sorted(patterns, key=len, reverse=True)), "***REDACTED***", text)
def _rate_limited_get(
self,
url: str,
params: Optional[dict] = None,
quiet: bool = False,
*,
_json_retry: bool = True,
) -> Optional[dict]:
"""Make a rate-limited GET request with budget enforcement.
Raises:
ApiCallBudgetExceeded: When api_calls_made >= max_api_calls
"""
if self.api_calls_made >= self.max_api_calls:
raise ApiCallBudgetExceeded(
f"API call budget exhausted: {self.api_calls_made}/{self.max_api_calls} calls used"
)
self._last_error = None
if self.rate_limit_reached:
self._last_error = "daily rate limit already reached"
return None
if params is None:
params = {}
elapsed = time.time() - self.last_call_time
if elapsed < self.RATE_LIMIT_DELAY:
time.sleep(self.RATE_LIMIT_DELAY - elapsed)
try:
response = self.session.get(url, params=params, timeout=30)
self.last_call_time = time.time()
self.api_calls_made += 1
if response.status_code == 200:
self.retry_count = 0
try:
data = response.json()
except ValueError as e:
# Malformed 200 body: retry once per call chain, then fail
# soft with a redacted reason instead of raising. The retry
# adds no backoff beyond the normal rate-limit delay and
# costs one extra API call (counts against any budget).
msg = self._redact_secret(f"invalid JSON response: {e}")
if _json_retry:
if not quiet:
print(f"WARNING: {msg}; retrying once", file=sys.stderr)
return self._rate_limited_get(url, params, quiet=quiet, _json_retry=False)
self._last_error = msg
if not quiet:
print(f"ERROR: {msg}", file=sys.stderr)
return None
return data
elif response.status_code == 429:
self.retry_count += 1
if self.retry_count <= self.max_retries:
print("WARNING: Rate limit exceeded. Waiting 60 seconds...", file=sys.stderr)
time.sleep(60)
return self._rate_limited_get(url, params, quiet=quiet, _json_retry=_json_retry)
else:
self._last_error = "HTTP 429 (daily rate limit)"
print("ERROR: Daily API rate limit reached.", file=sys.stderr)
self.rate_limit_reached = True
return None
else:
body = self._redact_secret(response.text)[:200]
msg = self._redact_secret(f"HTTP {response.status_code} - {body}")
self._last_error = msg
if not quiet:
print(
f"ERROR: API request failed: {msg}",
file=sys.stderr,
)
return None
except requests.exceptions.RequestException as e:
redacted = self._redact_secret(str(e))
self._last_error = f"request exception: {redacted}"
print(f"ERROR: Request exception: {redacted}", file=sys.stderr)
return None
def _request_with_fallback(self, endpoint_key, symbols_str, extra_params=None):
"""Try stable endpoint first, fall back to v3 for legacy users.
Returns parsed JSON in v3-compatible shape, or None if all fail.
Non-last endpoints are called with quiet=True so the user isn't
alarmed by an expected stable failure when v3 will catch it — but
when a non-last endpoint DOES fail, a WARN line is emitted explaining
why we're falling back. Otherwise users only see the (often misleading)
last-endpoint error and have no clue what really went wrong.
"""
params = dict(extra_params) if extra_params else {}
endpoints = _FMP_ENDPOINTS[endpoint_key]
is_single = "," not in symbols_str
for i, (base_url, url_builder) in enumerate(endpoints):
# Circuit breaker: skip endpoints with too many consecutive failures
if base_url in self._disabled_endpoints:
continue
url, final_params = url_builder(base_url, symbols_str, dict(params))
is_last = i == len(endpoints) - 1
data = self._rate_limited_get(url, final_params, quiet=not is_last)
if not data: # falsy (None, [], {}) — try next endpoint
self._record_endpoint_failure(base_url)
self._warn_fallback(base_url, is_last, self._last_error)
continue
# Normalize new stable EOD flat-list shape to v3-compatible dict.
# No-op for v3 dict / historicalStockList responses.
# `timeseries` (original request) is passed as `limit` so the
# EOD endpoint's full-history response is truncated to the
# legacy "most-recent N rows" contract.
if endpoint_key == "historical":
limit = params.get("timeseries") if isinstance(params, dict) else None
data = _normalize_eod_flat_list(data, symbols_str, limit=limit)
if not data:
self._record_endpoint_failure(base_url)
self._warn_fallback(
base_url,
is_last,
f"response had no rows matching '{symbols_str}'",
)
continue
# Shape validation: reject truthy-but-wrong-shape responses
valid = True
shape_issue: Optional[str] = None
if endpoint_key == "historical":
if not isinstance(data, dict):
valid = False
shape_issue = "expected dict"
elif "historicalStockList" in data:
# stable batch format -> v3 single format (exact match only)
norm = symbols_str.replace("-", ".")
found = None
for entry in data["historicalStockList"]:
if entry.get("symbol", "").replace("-", ".") == norm:
found = {
"symbol": entry.get("symbol"),
"historical": entry.get("historical", []),
}
break
if found:
self._endpoint_failures[base_url] = 0
return found
valid = False
shape_issue = f"'{symbols_str}' not in historicalStockList"
elif "historical" not in data:
valid = False
shape_issue = "missing 'historical' key"
elif is_single and data.get("symbol"):
if data["symbol"].replace("-", ".") != symbols_str.replace("-", "."):
valid = False
shape_issue = (
f"response symbol '{data['symbol']}' != requested '{symbols_str}'"
)
if valid:
self._endpoint_failures[base_url] = 0
return data
self._record_endpoint_failure(base_url)
self._warn_fallback(base_url, is_last, shape_issue or "unexpected response shape")
return None
def _warn_fallback(self, base_url: str, is_last: bool, reason: Optional[str]) -> None:
"""Emit a WARN line so users see why a non-last endpoint failed and the
client is falling back. No-op when the failing endpoint is the last one
(its error was already printed by _rate_limited_get with quiet=False)."""
if is_last or not reason:
return
print(
f"WARN: {base_url} failed ({reason}); falling back to next endpoint",
file=sys.stderr,
)
def _record_endpoint_failure(self, base_url: str) -> None:
"""Track consecutive failures and disable endpoint after threshold."""
failures = self._endpoint_failures.get(base_url, 0) + 1
self._endpoint_failures[base_url] = failures
if failures >= self._ENDPOINT_FAILURE_THRESHOLD:
self._disabled_endpoints.add(base_url)
# Stockbee primitives (#495 Batch C): stable-only, no v3 fallback and no
# circuit breaker -- an empty row list for a delisted/new ticker is data,
# not an endpoint failure, so one bad streak never disables the scan.
def _stable_rows(self, name: str, params: dict, quiet: bool = False):
data = self._rate_limited_get(f"{self.STABLE_URL}/{name}", params=params, quiet=quiet)
if isinstance(data, list):
return data
if data is not None:
# HTTP 200 with a non-list body (e.g. {"Error Message": ...}) is a
# provider-side failure, not an empty answer.
self._last_error = "unexpected non-list response"
if not quiet:
print("ERROR: FMP returned an unexpected non-list response", file=sys.stderr)
return None
def fetch_screener_rows(self, params: dict, quiet: bool = False) -> Optional[list]:
"""Raw /stable/company-screener rows; None on transport/HTTP failure."""
return self._stable_rows("company-screener", dict(params), quiet=quiet)
def fetch_eod_rows(
self, symbol: str, days: int, pad_days: int = 10, quiet: bool = False
) -> Optional[list]:
"""Raw /stable/historical-price-eod/full rows over an ET calendar window.
The window spans ``days * 2 + pad_days`` calendar days so ``days``
trading sessions fit. Returns None on transport/HTTP failure and []
when the provider answered without rows.
"""
today = _et_today()
params = {
"symbol": symbol,
"from": (today - timedelta(days=int(days) * 2 + int(pad_days))).isoformat(),
"to": today.isoformat(),
}
return self._stable_rows("historical-price-eod/full", params, quiet=quiet)
def fetch_quote_row(self, symbol: str, quiet: bool = True) -> Optional[dict]:
"""First /stable/quote row for one symbol, or None."""
rows = self._stable_rows("quote", {"symbol": symbol}, quiet=quiet)
if rows and isinstance(rows[0], dict):
return rows[0]
return None
def et_today(self):
"""America/New_York calendar date used for every query window."""
return _et_today()
def get_historical_prices(self, symbol: str, days: int = 80) -> Optional[list[dict]]:
"""Fetch historical daily OHLCV data for a symbol.
Args:
symbol: Ticker symbol
days: Number of trading days to fetch (default: 80)
Returns:
List of price dicts (most-recent-first) or None on failure.
"""
cache_key = f"prices_{symbol}_{days}"
if cache_key in self.cache:
return self.cache[cache_key]
data = self._request_with_fallback("historical", symbol, {"timeseries": days})
if data and "historical" in data:
result = data["historical"]
self.cache[cache_key] = result
return result
return None
def get_api_stats(self) -> dict:
"""Return API usage statistics."""
return {
"cache_entries": len(self.cache),
"api_calls_made": self.api_calls_made,
"max_api_calls": self.max_api_calls,
"rate_limit_reached": self.rate_limit_reached,
"budget_remaining": max(0, self.max_api_calls - self.api_calls_made),
}
scripts/tests/test_build_model_book_coverage.py
"""Additional coverage tests for build_model_book (Issue #293). No network access."""
import json
import sys
from datetime import date, timedelta
from pathlib import Path
from types import SimpleNamespace
import pytest
SCRIPT_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(SCRIPT_DIR))
import build_model_book as bmb # noqa: E402
from build_model_book import Bar # noqa: E402
def cand(**over):
base = {
"symbol": "TEST",
"date": "2026-06-20",
"state": "ACTIONABLE_DAY1",
"rating": "A",
"primary_trigger": "4pct_breakout",
"entry_reference": 100.0,
"stop_reference": 95.0,
}
base.update(over)
return base
def bars_up(start="2026-06-22", n=6, base=100.0):
out = []
d = date.fromisoformat(start)
for i in range(n):
px = base + i
out.append(Bar(d.isoformat(), px, px + 1, px - 0.5, px + 0.5, 1000))
d += timedelta(days=1)
return out
# ---------------------------------------------------------------- FMP client
class FakeResponse:
def __init__(self, status_code=200, payload=None, text=""):
self.status_code = status_code
self._payload = payload
self.text = text
def json(self):
if isinstance(self._payload, Exception):
raise self._payload
return self._payload
class FakeSession:
def __init__(self, responses):
self.headers = {}
self.responses = list(responses)
self.calls = []
def get(self, url, params=None, timeout=None):
self.calls.append((url, params, timeout))
return self.responses.pop(0)
def make_client(monkeypatch, responses, **kwargs):
"""Build the adapter with the real shared client but a fake HTTP session."""
session = FakeSession(responses)
monkeypatch.setattr(bmb, "requests", SimpleNamespace(Session=lambda: session))
shared = bmb._load_shared_fmp()
monkeypatch.setattr(shared.time, "sleep", lambda s: None)
client = bmb.FMPClient(api_key="k", **kwargs)
client.session = session
return client, session
def test_client_requires_requests(monkeypatch):
monkeypatch.setattr(bmb, "requests", None)
with pytest.raises(RuntimeError, match="requests is not installed"):
bmb.FMPClient(api_key="k")
def test_client_requires_api_key(monkeypatch):
monkeypatch.setattr(bmb, "requests", SimpleNamespace(Session=lambda: FakeSession([])))
monkeypatch.delenv("FMP_API_KEY", raising=False)
with pytest.raises(ValueError, match="FMP API key required"):
bmb.FMPClient()
def test_client_reads_key_from_env_and_sets_header(monkeypatch):
monkeypatch.setenv("FMP_API_KEY", "envkey")
client = bmb.FMPClient()
assert client.api_key == "envkey" # pragma: allowlist secret
assert client.session.headers["apikey"] == "envkey" # pragma: allowlist secret
def test_explicit_api_key_beats_env(monkeypatch):
monkeypatch.setenv("FMP_API_KEY", "envkey")
client = bmb.FMPClient(api_key="argkey")
assert client.api_key == "argkey" # pragma: allowlist secret
assert client.session.headers["apikey"] == "argkey" # pragma: allowlist secret
def test_shared_module_is_private_and_cached():
first = bmb._load_shared_fmp()
assert first is bmb._load_shared_fmp()
assert sys.modules[bmb._SHARED_FMP_MODULE] is first
assert bmb._SHARED_FMP_MODULE != "fmp_client"
def test_get_stable_success_and_stats(monkeypatch):
rows = [
{"date": "2026-06-22", "open": 1, "high": 2, "low": 0.5, "close": 1.5, "volume": 10},
{"date": "2026-06-23", "open": 1, "high": 2, "low": 0.5, "close": 1.8, "volume": 10},
]
client, session = make_client(monkeypatch, [FakeResponse(200, rows)])
out = client.get_historical_prices("brk.b", days=1)
assert [r["date"] for r in out] == ["2026-06-23"]
url, params, timeout = session.calls[0]
assert url.endswith("/stable/historical-price-eod/full")
assert params["symbol"] == "BRK-B"
assert timeout == 30
span = date.fromisoformat(params["to"]) - date.fromisoformat(params["from"])
assert span.days == 1 * 2 + 10
assert client.stats() == {
"api_calls_made": 1,
"max_api_calls": 200,
"budget_remaining": 199,
"rate_limit_reached": False,
"failed_requests": 0,
}
def test_get_historical_prices_caches(monkeypatch):
rows = [{"date": "2026-06-22", "close": 5}]
client, session = make_client(monkeypatch, [FakeResponse(200, rows)])
first = client.get_historical_prices("AAA")
second = client.get_historical_prices("AAA")
assert first is second
assert len(session.calls) == 1
def test_empty_rows_not_a_failure_and_no_v3(monkeypatch):
client, session = make_client(monkeypatch, [FakeResponse(200, [])])
assert client.get_historical_prices("AAA") == []
assert len(session.calls) == 1
assert client.stats()["failed_requests"] == 0
def test_http_error_counts_failure_and_never_requests_v3(monkeypatch, capsys):
client, session = make_client(monkeypatch, [FakeResponse(402, None, "pay")])
assert client.get_historical_prices("AAA") == []
assert len(session.calls) == 1
assert all("/api/v3/" not in call[0] for call in session.calls)
assert client.stats()["failed_requests"] == 1
# failures are not cached: a later call retries
session.responses.append(FakeResponse(200, [{"date": "2026-06-22", "close": 5}]))
assert len(client.get_historical_prices("AAA")) == 1
def test_non_list_body_counts_failure(monkeypatch):
client, _ = make_client(monkeypatch, [FakeResponse(200, {"Error Message": "bad"})])
assert client.get_historical_prices("AAA") == []
assert client.stats()["failed_requests"] == 1
def test_malformed_json_retries_once_then_succeeds(monkeypatch):
rows = [{"date": "2026-06-22", "close": 5}]
client, session = make_client(
monkeypatch, [FakeResponse(200, ValueError("boom")), FakeResponse(200, rows)]
)
assert len(client.get_historical_prices("AAA")) == 1
assert len(session.calls) == 2
assert client.stats()["failed_requests"] == 0
def test_malformed_json_twice_fails_soft(monkeypatch):
client, _ = make_client(
monkeypatch, [FakeResponse(200, ValueError("a")), FakeResponse(200, ValueError("b"))]
)
assert client.get_historical_prices("AAA") == []
assert client.stats()["failed_requests"] == 1
def test_rate_limit_sets_flag_and_counts_failure(monkeypatch):
client, session = make_client(monkeypatch, [FakeResponse(429), FakeResponse(429)])
assert client.get_historical_prices("AAA") == []
assert client.rate_limit_reached is True
stats = client.stats()
assert stats["rate_limit_reached"] is True
assert stats["failed_requests"] == 1
assert len(session.calls) == 2
def test_error_body_key_is_redacted(monkeypatch, capsys):
secret = "SECRETKEY123" # pragma: allowlist secret
client, _ = make_client(monkeypatch, [FakeResponse(500, None, f"bad {secret} {secret}")])
client._client.api_key = secret
client.get_historical_prices("AAA")
captured = capsys.readouterr()
assert secret not in captured.err
assert "***REDACTED***" in captured.err
assert secret not in captured.out
def test_budget_exhausted_raises_local_exception(monkeypatch):
client, session = make_client(monkeypatch, [], max_api_calls=0)
with pytest.raises(bmb.ApiCallBudgetExceeded, match="0/0"):
client.get_historical_prices("AAA")
assert session.calls == []
def test_budget_exhausted_mid_loop(monkeypatch):
client, _ = make_client(monkeypatch, [FakeResponse(200, [])], max_api_calls=1)
client.get_historical_prices("AAA")
with pytest.raises(bmb.ApiCallBudgetExceeded):
client.get_historical_prices("BBB")
def test_no_circuit_breaker_after_empty_symbols(monkeypatch):
rows = [{"date": "2026-06-22", "close": 5}]
responses = [FakeResponse(200, [])] * 5 + [FakeResponse(200, rows)]
client, session = make_client(monkeypatch, responses)
for i in range(5):
assert client.get_historical_prices(f"E{i}") == []
assert len(client.get_historical_prices("GOOD")) == 1
assert len(session.calls) == 6
def test_delegated_attributes_roundtrip(monkeypatch):
client, _ = make_client(monkeypatch, [])
client.api_calls_made = 3
client.max_api_calls = 9
client.last_call_time = 12.5
client.rate_limit_reached = True
assert (client._client.api_calls_made, client._client.max_api_calls) == (3, 9)
assert client._client.last_call_time == 12.5
assert client._client.rate_limit_reached is True
assert client.stats()["budget_remaining"] == 6
# ------------------------------------------------------------- tiny helpers
@pytest.mark.parametrize(
"value,expected",
[
(None, 0.0),
("", 0.0),
("abc", 0.0),
(float("nan"), 0.0),
(float("inf"), 0.0),
("2.5", 2.5),
(3, 3.0),
],
)
def test_to_float(value, expected):
assert bmb.to_float(value) == expected
def test_to_float_custom_default():
assert bmb.to_float(None, default=-1.0) == -1.0
@pytest.mark.parametrize(
"value,expected", [(None, 0), ("", 0), ("x", 0), ("7.9", 7), (4, 4), ([], 0)]
)
def test_to_int(value, expected):
assert bmb.to_int(value) == expected
def test_normalize_symbol():
assert bmb.normalize_symbol(" brk.b ") == "BRK-B"
assert bmb.normalize_symbol(None) == ""
def test_parse_date_truncates_timestamp():
assert bmb.parse_date("2026-06-20T10:00:00") == date(2026, 6, 20)
# ----------------------------------------------------- normalize_price_bars
def test_normalize_bars_unsupported_payload():
assert bmb.normalize_price_bars("junk") == []
assert bmb.normalize_price_bars(None) == []
assert bmb.normalize_price_bars({"other": 1}) == []
def test_normalize_bars_historical_and_bars_keys():
rows = [{"date": "2026-06-22", "close": 10}]
assert bmb.normalize_price_bars({"historical": rows})[0].close == 10.0
assert bmb.normalize_price_bars({"bars": rows})[0].close == 10.0
assert bmb.normalize_price_bars({"historical": None}) == []
def test_normalize_bars_filters_and_fills_defaults():
rows = [
"not-a-dict",
{"date": "bad", "close": 10},
{"date": "2026-06-22", "close": 0},
{"date": "2026-06-22", "symbol": "OTHER", "close": 50},
{"datetime": "2026-06-23 09:30:00", "close": 20},
{"date": "2026-06-24", "close": None, "adjClose": 30},
{"date": "2026-06-25", "open": 10, "high": 0, "low": 0, "close": 12, "symbol": "tst"},
]
bars = bmb.normalize_price_bars(rows, symbol="TST")
by_date = {b.date: b for b in bars}
assert sorted(by_date) == ["2026-06-23", "2026-06-24", "2026-06-25"]
b23 = by_date["2026-06-23"]
assert (b23.open, b23.high, b23.low, b23.close) == (20.0, 20.0, 20.0, 20.0)
# close falls back to adjClose; open/high/low defaults were derived from missing close
assert by_date["2026-06-24"].close == 30.0
b25 = by_date["2026-06-25"]
assert (b25.high, b25.low) == (12.0, 12.0) # non-positive high/low -> close
def test_normalize_bars_dedups_by_date_keeping_last():
rows = [{"date": "2026-06-22", "close": 10}, {"date": "2026-06-22", "close": 11}]
bars = bmb.normalize_price_bars(rows)
assert len(bars) == 1
assert bars[0].close == 11.0
# ----------------------------------------------------------- load_prices_json
def test_load_prices_json_symbol_mapping(tmp_path):
p = tmp_path / "p.json"
p.write_text(
json.dumps(
{
"aaa": [{"date": "2026-06-22", "close": 5}],
"EMPTY": [],
"BAD": "x",
}
),
encoding="utf-8",
)
prices = bmb.load_prices_json(p)
assert list(prices) == ["AAA"]
assert prices["AAA"][0].close == 5.0
def test_load_prices_json_flat_list(tmp_path):
p = tmp_path / "p.json"
p.write_text(
json.dumps(
[
{"symbol": "aaa", "date": "2026-06-22", "close": 5},
{"ticker": "bbb", "date": "2026-06-22", "close": 6},
{"date": "2026-06-22", "close": 7},
"junk",
{"symbol": "ccc", "date": "bad", "close": 8},
]
),
encoding="utf-8",
)
prices = bmb.load_prices_json(p)
assert sorted(prices) == ["AAA", "BBB"]
assert prices["BBB"][0].close == 6.0
def test_load_prices_json_scalar_returns_empty(tmp_path):
p = tmp_path / "p.json"
p.write_text("42", encoding="utf-8")
assert bmb.load_prices_json(p) == {}
# -------------------------------------------------------- model book storage
def test_load_model_book_missing_file(tmp_path):
assert bmb.load_model_book(tmp_path / "nope.jsonl") == []
def test_load_model_book_skips_blank_and_non_dict(tmp_path):
p = tmp_path / "b.jsonl"
p.write_text('{"a": 1}\n\n[1, 2]\n{"b": 2}\n', encoding="utf-8")
assert bmb.load_model_book(p) == [{"a": 1}, {"b": 2}]
def test_load_model_book_invalid_json_reports_line(tmp_path):
p = tmp_path / "b.jsonl"
p.write_text('{"a": 1}\n{oops\n', encoding="utf-8")
with pytest.raises(ValueError, match=r"b\.jsonl:2"):
bmb.load_model_book(p)
def test_save_model_book_sorts_and_creates_dirs(tmp_path):
p = tmp_path / "deep" / "dir" / "b.jsonl"
recs = [
{"setup_date": "2026-06-21", "symbol": "B", "record_id": "3"},
{"setup_date": "2026-06-20", "symbol": "B", "record_id": "2"},
{"setup_date": "2026-06-20", "symbol": "A", "record_id": "1", "note": "日本語"},
]
bmb.save_model_book(p, recs)
text = p.read_text(encoding="utf-8")
assert "日本語" in text
assert [r["record_id"] for r in bmb.load_model_book(p)] == ["1", "2", "3"]
# ----------------------------------------------------------- upsert_records
def test_upsert_inserts_new_and_ignores_existing_without_id():
existing = [{"symbol": "X"}]
new = [{"record_id": "n1", "v": 1}]
records, stats = bmb.upsert_records(existing, new)
assert stats == {"inserted": 1, "updated": 0, "total": 1}
assert records == [{"record_id": "n1", "v": 1}]
def test_upsert_preserves_human_annotations():
prev = {"record_id": "r", "human_label": "good", "human_decision": "take", "human_notes": "n"}
fresh = {
"record_id": "r",
"human_label": None,
"human_decision": "unknown",
"human_notes": "",
"setup_score": 9,
}
records, stats = bmb.upsert_records([prev], [fresh])
assert stats == {"inserted": 0, "updated": 1, "total": 1}
assert records[0]["human_label"] == "good"
assert records[0]["human_decision"] == "take"
assert records[0]["human_notes"] == "n"
assert records[0]["setup_score"] == 9
def test_upsert_keeps_matured_outcomes_over_pending_placeholder():
prev = {
"record_id": "r",
"outcomes": {"5d": {"matured": True}},
"overall_outcome": "WORKED",
"matured": True,
"last_outcome_update_at": "t",
}
fresh = {"record_id": "r", "outcomes": {}, "overall_outcome": "PENDING", "matured": False}
records, _ = bmb.upsert_records([prev], [fresh])
assert records[0]["overall_outcome"] == "WORKED"
assert records[0]["matured"] is True
assert records[0]["last_outcome_update_at"] == "t"
assert records[0]["outcomes"] == {"5d": {"matured": True}}
def test_upsert_new_outcomes_override_previous():
prev = {"record_id": "r", "outcomes": {"5d": {"matured": True}}, "overall_outcome": "WORKED"}
fresh = {
"record_id": "r",
"outcomes": {"5d": {"matured": True, "x": 1}},
"overall_outcome": "X",
}
records, _ = bmb.upsert_records([prev], [fresh])
assert records[0]["overall_outcome"] == "X"
# ------------------------------------------------- load_screener_candidates
def test_load_screener_rejects_non_report(tmp_path):
p = tmp_path / "bad.json"
p.write_text(json.dumps({"nothing": 1}), encoding="utf-8")
with pytest.raises(ValueError, match="does not look like"):
bmb.load_screener_candidates([p])
p.write_text(json.dumps([1, 2]), encoding="utf-8")
with pytest.raises(ValueError, match="does not look like"):
bmb.load_screener_candidates([p])
def test_load_screener_filters_rejects_and_attaches_source(tmp_path):
p = tmp_path / "r.json"
p.write_text(
json.dumps(
{
"metadata": {"src": "x"},
"candidates": [
cand(symbol="OK"),
cand(symbol="R1", state="reject"),
cand(symbol="R2", state="REJECTED"),
cand(symbol="R3", state="Invalid"),
"junk",
],
}
),
encoding="utf-8",
)
kept = bmb.load_screener_candidates([p])
assert [c["symbol"] for c in kept] == ["OK"]
assert kept[0]["_source_report"] == str(p)
assert kept[0]["_source_metadata"] == {"src": "x"}
allc = bmb.load_screener_candidates([p], include_rejects=True)
assert [c["symbol"] for c in allc] == ["OK", "R1", "R2", "R3"]
def test_load_screener_non_dict_metadata_becomes_empty(tmp_path):
p = tmp_path / "r.json"
p.write_text(json.dumps({"metadata": "str", "candidates": [cand()]}), encoding="utf-8")
assert bmb.load_screener_candidates([p])[0]["_source_metadata"] == {}
# --------------------------------------------------------- derive_setup_tags
def test_derive_tags_empty_candidate():
assert bmb.derive_setup_tags({}) == ["weak_close_location"]
def test_derive_tags_low_side_branches():
tags = bmb.derive_setup_tags(
{
"trigger_tags": ["t1", "", None],
"soft_failure_tags": ["s1", ""],
"reject_reasons": ["r1", None],
"rating": "A+",
"close_location_pct": 49.9,
"volume_ratio_1d": 1.0,
"base_width_pct": 20,
"prior_base_days": 31,
"risk_pct_to_stop": 6,
"recent_4pct_breakdown": True,
"prior_up_streak": 3,
}
)
assert tags == sorted(
[
"t1",
"soft_s1",
"reject_r1",
"rating_Aplus",
"weak_close_location",
"thin_volume_confirmation",
"wide_base",
"long_base",
"wide_risk",
"recent_4pct_breakdown",
"three_days_up_before_trigger",
]
)
def test_derive_tags_minus_rating_and_compact_risk():
tags = bmb.derive_setup_tags(
{"rating": "B-", "risk_pct_to_stop": 2.0, "close_location_pct": 60}
)
assert tags == ["compact_risk", "rating_Bminus"]
# ------------------------------------------------------------- make_record
def test_make_record_missing_symbol_or_date_raises():
with pytest.raises(ValueError, match="missing symbol or setup date"):
bmb.make_record({"date": "2026-06-20"})
with pytest.raises(ValueError, match="missing symbol or setup date"):
bmb.make_record({"symbol": "A", "date": "2026"})
def test_make_record_fallbacks():
rec = bmb.make_record(
{"symbol": "a.b", "setup_date": "2026-06-20T00:00:00", "close": 50.0, "low": 48.0}
)
assert rec["symbol"] == "A-B"
assert rec["setup_date"] == "2026-06-20"
assert rec["primary_trigger"] == "unknown"
assert rec["record_id"] == "stockbee_mb:A-B:2026-06-20:unknown"
assert rec["entry_reference"] == 50.0
assert rec["stop_reference"] == 48.0
assert rec["trigger_tags"] == []
assert rec["human_decision"] == "unknown"
assert rec["source_report"] is None
def test_is_studyable_candidate_reasons():
assert bmb.is_studyable_candidate({"date": "2026-06-20"}) == (False, "missing_symbol")
assert bmb.is_studyable_candidate({"symbol": "A"}) == (False, "missing_setup_date")
assert bmb.is_studyable_candidate(cand()) == (True, None)
def test_make_studyable_records_counts_missing_symbol():
records, stats = bmb.make_studyable_records([cand(), {"date": "2026-06-20"}, {"symbol": "Z"}])
assert len(records) == 1
assert stats == {
"candidates_loaded": 3,
"studyable": 1,
"skipped": 2,
"skipped_missing_symbol": 1,
"skipped_missing_setup_date": 1,
}
# ------------------------------------------------------ outcome calculation
def test_window_outcome_missing_entry():
out = bmb.calculate_window_outcome({"setup_date": "2026-06-20"}, bars_up(), 3)
assert out == {"horizon_days": 3, "matured": False, "reason": "missing_entry"}
def test_window_outcome_not_enough_bars():
rec = bmb.make_record(cand())
out = bmb.calculate_window_outcome(rec, bars_up(n=2), 3)
assert out == {
"horizon_days": 3,
"matured": False,
"available_future_bars": 2,
"reason": "not_enough_future_bars",
}
def test_window_outcome_excludes_setup_day_and_no_stop():
rec = bmb.make_record(cand(stop_reference=0, entry_reference=100.0))
rec["stop_reference"] = 0.0
bars = [Bar("2026-06-20", 1, 500, 0.1, 100, 1)] + bars_up(n=3)
out = bmb.calculate_window_outcome(rec, bars, 3)
assert out["matured"] is True
assert out["close_date"] == "2026-06-24"
assert out["stop_hit"] is False
assert out["stop_hit_date"] is None
assert out["max_high"] == 103.0
assert out["min_low"] == 99.5
assert out["forward_return_pct"] == 2.5
assert out["mfe_pct"] == 3.0
assert out["mae_pct"] == -0.5
assert out["outcome_tag"] == "NEUTRAL"
@pytest.mark.parametrize(
"args,tag",
[
((1.0, 1.0, -1.0, True), "FAILED_STOP"),
((0.0, 12.0, 0.0, False), "STRONG_WINNER"),
((8.0, 1.0, 0.0, False), "STRONG_WINNER"),
((0.0, 6.0, 0.0, False), "WORKED"),
((4.0, 1.0, 0.0, False), "WORKED"),
((-2.0, 1.0, 0.0, False), "FAILED_FADE"),
((1.0, 1.0, -5.0, False), "CHOPPY_FAILURE"),
((2.0, 1.0, -5.0, False), "NEUTRAL"),
],
)
def test_classify_outcome_branches(args, tag):
assert bmb.classify_outcome(*args) == tag
def test_derive_overall_outcome():
assert bmb.derive_overall_outcome({}) == "PENDING"
assert bmb.derive_overall_outcome({"5d": {"matured": False}}) == "PENDING"
assert bmb.derive_overall_outcome({"5d": {"matured": True}}) == "UNKNOWN"
assert bmb.derive_overall_outcome({"3d": {"matured": True, "outcome_tag": "WORKED"}}, 3) == (
"WORKED"
)
def test_update_outcomes_missing_prices_pending_and_skip_matured():
missing = bmb.make_record(cand(symbol="NOPX"))
pending = bmb.make_record(cand(symbol="SHORT"))
done = bmb.make_record(cand(symbol="DONE"))
done["matured"] = True
records, stats = bmb.update_record_outcomes(
[missing, pending, done], {"SHORT": bars_up(n=2)}, horizons=(3, 5)
)
assert stats == {"updated": 1, "matured": 0, "still_pending": 1, "skipped_no_prices": 1}
assert missing["update_error"] == "missing_prices"
assert pending["overall_outcome"] == "PENDING"
assert pending["matured"] is False
assert pending["outcomes"]["5d"]["reason"] == "not_enough_future_bars"
assert "outcomes" in done and done["outcomes"] == {}
def test_update_outcomes_only_pending_false_reprocesses_matured():
rec = bmb.make_record(cand())
rec["matured"] = True
rec["overall_outcome"] = "WORKED"
_, stats = bmb.update_record_outcomes([rec], {"TEST": bars_up(n=6)}, only_pending=False)
assert stats["updated"] == 1
assert stats["matured"] == 1
assert rec["outcomes"]["5d"]["matured"] is True
assert rec["last_outcome_update_at"] == rec["updated_at"]
# ------------------------------------------------------------- summarizing
def test_values_for_metric_nested_and_non_dict_paths():
recs = [
{"a": {"b": 1.5}},
{"a": 3},
{"a": {"b": None}},
{"a": {"b": "x"}},
{},
{"a": {"b": 2}},
]
assert bmb.values_for_metric(recs, "a.b") == [1.5, 2.0]
def test_group_records_untagged_and_unknown():
recs = [
{"setup_tags": ["x", "y"], "rating": "A"},
{"setup_tags": [], "rating": None},
]
tags = bmb.group_records(recs, "setup_tags")
assert sorted(tags) == ["untagged", "x", "y"]
ratings = bmb.group_records(recs, "rating")
assert sorted(ratings) == ["A", "unknown"]
def test_summarize_group_no_matured():
assert bmb.summarize_group([{"matured": False}, {}]) == {
"sample_size": 0,
"pending": 2,
"win_rate_pct": None,
}
def make_matured(outcome, r5=None, mae=-1.0, stop=False, r3=None):
outcomes = {"5d": {"mae_pct": mae, "mfe_pct": 5.0, "stop_hit": stop}}
if r5 is not None:
outcomes["5d"]["forward_return_pct"] = r5
if r3 is not None:
outcomes["3d"] = {"forward_return_pct": r3}
return {"matured": True, "overall_outcome": outcome, "outcomes": outcomes}
def test_summarize_group_rates_and_averages():
recs = [
make_matured("STRONG_WINNER", r5=10, r3=6),
make_matured("WORKED", r5=4, r3=2),
make_matured("FAILED_STOP", r5=-6, stop=True, mae=-7),
make_matured("CHOPPY_FAILURE", r5=0, mae=-6),
{"matured": False},
]
s = bmb.summarize_group(recs)
assert s["sample_size"] == 4
assert s["pending"] == 1
assert s["win_rate_pct"] == 50.0
assert s["failure_rate_pct"] == 50.0
assert s["strong_winner_rate_pct"] == 25.0
assert s["stop_hit_rate_pct"] == 25.0
assert s["avg_3d_return_pct"] == 4.0
assert s["avg_5d_return_pct"] == 2.0
assert s["median_5d_return_pct"] == 2.0
assert s["avg_5d_mfe_pct"] == 5.0
assert s["avg_5d_mae_pct"] == -3.75
def test_summarize_group_missing_metrics_are_none():
s = bmb.summarize_group([{"matured": True, "overall_outcome": "NEUTRAL"}])
assert s["sample_size"] == 1
assert s["win_rate_pct"] == 0.0
assert s["avg_3d_return_pct"] is None
assert s["avg_5d_return_pct"] is None
assert s["median_5d_return_pct"] is None
assert s["avg_5d_mfe_pct"] is None
assert s["avg_5d_mae_pct"] is None
def test_is_promote_and_downgrade_thresholds():
good = {"sample_size": 5, "win_rate_pct": 60, "avg_5d_return_pct": 2, "avg_5d_mae_pct": -4.9}
assert bmb.is_promote_candidate(good) is True
assert bmb.is_promote_candidate({**good, "sample_size": 4}) is False
assert bmb.is_promote_candidate({**good, "avg_5d_mae_pct": -5}) is False
assert bmb.is_promote_candidate({}) is False
assert bmb.is_downgrade_candidate({"sample_size": 5, "failure_rate_pct": 50}) is True
assert bmb.is_downgrade_candidate({"sample_size": 5, "avg_5d_return_pct": -1}) is True
assert bmb.is_downgrade_candidate({"sample_size": 4, "failure_rate_pct": 90}) is False
assert bmb.is_downgrade_candidate({"sample_size": 5, "failure_rate_pct": 10}) is False
def test_summarize_model_book_downgrade_threshold_and_sorting():
recs = []
for _i in range(5):
r = make_matured("FAILED_FADE", r5=-3)
r["rating"] = "C"
recs.append(r)
for _i in range(3):
r = make_matured("WORKED", r5=5)
r["rating"] = "B"
recs.append(r)
for _i in range(2): # below min_sample -> dropped
r = make_matured("WORKED", r5=9)
r["rating"] = "A"
recs.append(r)
s = bmb.summarize_model_book(recs, ["rating"], min_sample=3)
assert s["total_records"] == 10
assert [row["group"] for row in s["groups"]["rating"]] == ["B", "C"]
assert len(s["rule_candidates"]) == 1
rule = s["rule_candidates"][0]
assert rule["action"] == "downgrade_or_filter"
assert rule["group"] == "C"
assert rule["evidence"]["sample_size"] == 5
def test_summarize_model_book_empty():
s = bmb.summarize_model_book([], ["rating"])
assert s["total_records"] == 0
assert s["overall"] == {"sample_size": 0, "pending": 0, "win_rate_pct": None}
assert s["groups"] == {"rating": []}
assert s["rule_candidates"] == []
# ------------------------------------------------------------------ reports
def test_write_json_report_creates_dir(tmp_path):
out = tmp_path / "a" / "b"
path = bmb.write_json_report({"k": "日本"}, out, "pre")
assert path.parent == out
assert path.name.startswith("pre_")
assert path.suffix == ".json"
assert json.loads(path.read_text(encoding="utf-8")) == {"k": "日本"}
def test_markdown_report_minimal(tmp_path):
path = bmb.write_markdown_report({"generated_at": "G"}, tmp_path / "m", "p")
text = path.read_text(encoding="utf-8")
assert text.startswith("# Stockbee Setup Fluency Report\n\nGenerated: G\n")
assert "## Reminder" in text
assert "## Ingest" not in text
assert "## Rule Candidates" not in text
def test_markdown_report_default_generated_stamp(tmp_path):
path = bmb.write_markdown_report({"title": "T"}, tmp_path, "p")
lines = path.read_text(encoding="utf-8").splitlines()
assert lines[0] == "# T"
assert lines[2].startswith("Generated: 20")
def test_markdown_report_ingest_and_update_sections(tmp_path):
payload = {
"ingest": {
"candidates_loaded": 4,
"studyable": 2,
"skipped": 2,
"skipped_missing_setup_date": 1,
"skipped_missing_symbol": 1,
"inserted": 1,
"updated": 1,
"total": 9,
},
"update": {"updated": 3, "matured": 2, "still_pending": 1, "skipped_no_prices": 5},
}
text = bmb.write_markdown_report(payload, tmp_path, "p").read_text(encoding="utf-8")
assert "- Candidates loaded: 4" in text
assert "- Skipped missing setup date: 1" in text
assert "- Skipped missing symbol: 1" in text
assert "- Total model-book records: 9" in text
assert "## Outcome Update" in text
assert "- Matured: 2" in text
assert "- Missing prices: 5" in text
def test_markdown_report_ingest_omits_zero_skip_lines(tmp_path):
text = bmb.write_markdown_report({"ingest": {"studyable": 1}}, tmp_path, "p").read_text(
encoding="utf-8"
)
assert "Skipped missing" not in text
assert "- Skipped: 0" in text
def test_markdown_report_summary_groups_and_rules(tmp_path):
row = {
"group": "A",
"sample_size": 6,
"win_rate_pct": 80.0,
"failure_rate_pct": 10.0,
"avg_5d_return_pct": 3.0,
"avg_5d_mfe_pct": 7.0,
"avg_5d_mae_pct": -2.0,
}
payload = {
"total_records": 6,
"matured_records": 6,
"pending_records": 0,
"overall": {
"sample_size": 6,
"win_rate_pct": 80.0,
"avg_5d_return_pct": 3.0,
"avg_5d_mfe_pct": 7.0,
"avg_5d_mae_pct": -2.0,
},
"groups": {"rating": [row], "primary_trigger": []},
"rule_candidates": [
{
"action": "promote",
"group_by": "rating",
"group": "A",
"reason": "why",
"evidence": row,
}
],
}
text = bmb.write_markdown_report(payload, tmp_path, "p").read_text(encoding="utf-8")
assert "- Win rate: 80.0%" in text
assert "- Avg 5d MFE / MAE: 7.0% / -2.0%" in text
assert "## Cohorts by rating" in text
assert "| A | 6 | 80.0 | 10.0 | 3.0 | 7.0 | -2.0 |" in text
assert "## Cohorts by primary_trigger\n\nNo cohorts met the sample threshold." in text
assert "- **promote** `rating=A`: why N=6, Win=80.0%, Avg5d=3.0%." in text
def test_markdown_report_overall_without_samples(tmp_path):
payload = {"overall": {"sample_size": 0}, "total_records": 2, "pending_records": 2}
text = bmb.write_markdown_report(payload, tmp_path, "p").read_text(encoding="utf-8")
assert "- Pending records: 2" in text
assert "Win rate" not in text
# ----------------------------------------------------------- parse_horizons
def test_parse_horizons_dedups_and_sorts():
assert bmb.parse_horizons("5, 3,,3") == (3, 5)
def test_parse_horizons_errors():
with pytest.raises(ValueError, match="positive integers"):
bmb.parse_horizons("0")
with pytest.raises(ValueError, match="positive integers"):
bmb.parse_horizons("-2,3")
with pytest.raises(ValueError, match="at least one horizon"):
bmb.parse_horizons(" , ")
with pytest.raises(ValueError):
bmb.parse_horizons("abc")
# ----------------------------------------------------------------- commands
def write_screener(path, candidates):
path.write_text(json.dumps({"metadata": {}, "candidates": candidates}), encoding="utf-8")
return path
def test_main_ingest_end_to_end(tmp_path, capsys):
screener = write_screener(
tmp_path / "s.json", [cand(), cand(symbol="REJ", state="REJECT"), {"symbol": "NODATE"}]
)
book = tmp_path / "state" / "book.jsonl"
out = tmp_path / "out"
rc = bmb.main(
[
"ingest",
"--screener-json",
str(screener),
"--model-book",
str(book),
"--output-dir",
str(out),
]
)
assert rc == 0
stdout = capsys.readouterr().out
assert "Ingested 1 studyable candidates" in stdout
assert "(1 skipped)" in stdout
records = bmb.load_model_book(book)
assert [r["symbol"] for r in records] == ["TEST"]
assert records[0]["source_report"] == str(screener)
reports = sorted(p.name for p in out.iterdir())
assert len(reports) == 2
jpath = next(p for p in out.iterdir() if p.suffix == ".json")
payload = json.loads(jpath.read_text(encoding="utf-8"))
assert payload["ingest"]["inserted"] == 1
assert payload["candidates_loaded"] == 2
assert payload["model_book"] == str(book)
def test_main_ingest_include_rejects_and_reingest_updates(tmp_path, capsys):
screener = write_screener(tmp_path / "s.json", [cand(), cand(symbol="REJ", state="REJECT")])
book = tmp_path / "book.jsonl"
argv = ["ingest", "--screener-json", str(screener), "--model-book", str(book)]
argv += ["--output-dir", str(tmp_path / "o"), "--include-rejects"]
assert bmb.main(argv) == 0
assert bmb.main(argv) == 0
capsys.readouterr()
records = bmb.load_model_book(book)
assert sorted(r["symbol"] for r in records) == ["REJ", "TEST"]
jsons = sorted((tmp_path / "o").glob("*.json"))
assert jsons
last = max(jsons, key=lambda p: p.stat().st_mtime_ns)
assert json.loads(last.read_text(encoding="utf-8"))["ingest"]["total"] == 2
def test_main_update_no_records(tmp_path, capsys):
rc = bmb.main(["update", "--model-book", str(tmp_path / "none.jsonl")])
assert rc == 0
assert "No records found in" in capsys.readouterr().out
assert not (tmp_path / "none.jsonl").exists()
def test_main_update_with_prices_json(tmp_path, capsys):
book = tmp_path / "book.jsonl"
bmb.save_model_book(book, [bmb.make_record(cand())])
prices = tmp_path / "prices.json"
rows = [
{"date": b.date, "open": b.open, "high": b.high, "low": b.low, "close": b.close}
for b in bars_up(n=6)
]
prices.write_text(json.dumps({"TEST": rows}), encoding="utf-8")
out = tmp_path / "out"
rc = bmb.main(
[
"update",
"--model-book",
str(book),
"--prices-json",
str(prices),
"--horizons",
"5,3",
"--output-dir",
str(out),
]
)
assert rc == 0
assert "Updated outcomes in" in capsys.readouterr().out
rec = bmb.load_model_book(book)[0]
assert rec["matured"] is True
assert set(rec["outcomes"]) == {"3d", "5d"}
jpath = next(p for p in out.iterdir() if p.suffix == ".json")
payload = json.loads(jpath.read_text(encoding="utf-8"))
assert payload["horizons"] == [3, 5]
assert payload["prices_source"] == str(prices)
assert payload["api_stats"] is None
assert payload["update"]["matured"] == 1
def test_main_update_no_only_pending_flag(tmp_path, capsys):
book = tmp_path / "book.jsonl"
rec = bmb.make_record(cand())
rec["matured"] = True
rec["overall_outcome"] = "WORKED"
bmb.save_model_book(book, [rec])
prices = tmp_path / "prices.json"
rows = [{"date": b.date, "close": b.close} for b in bars_up(n=6)]
prices.write_text(json.dumps({"TEST": rows}), encoding="utf-8")
argv = ["update", "--model-book", str(book), "--prices-json", str(prices)]
argv += ["--output-dir", str(tmp_path / "o"), "--no-only-pending"]
assert bmb.main(argv) == 0
capsys.readouterr()
assert bmb.load_model_book(book)[0]["outcomes"]["5d"]["matured"] is True
def install_fake_session(monkeypatch, session):
"""Route the adapter's shared client through a fake HTTP session."""
shared = bmb._load_shared_fmp()
fake_requests = SimpleNamespace(Session=lambda: session, exceptions=shared.requests.exceptions)
monkeypatch.setattr(bmb, "requests", fake_requests)
monkeypatch.setattr(shared, "requests", fake_requests)
monkeypatch.setattr(shared.time, "sleep", lambda s: None)
monkeypatch.setenv("FMP_API_KEY", "k")
def fmp_rows():
return [
{"date": b.date, "open": b.open, "high": b.high, "low": b.low, "close": b.close}
for b in bars_up(n=6)
]
def update_argv(book, out_dir, max_calls="5"):
argv = ["update", "--model-book", str(book), "--history-days", "10"]
return argv + ["--max-api-calls", max_calls, "--output-dir", str(out_dir)]
def test_main_update_via_fmp_uses_fake_client(tmp_path, monkeypatch, capsys):
book = tmp_path / "book.jsonl"
bmb.save_model_book(
book,
[bmb.make_record(cand(symbol="aaa")), bmb.make_record(cand(symbol="bbb"))],
)
session = FakeSession([FakeResponse(200, fmp_rows()), FakeResponse(200, [])])
install_fake_session(monkeypatch, session)
assert bmb.main(update_argv(book, tmp_path / "o")) == 0
capsys.readouterr()
recs = {r["symbol"]: r for r in bmb.load_model_book(book)}
assert recs["AAA"]["matured"] is True
assert recs["BBB"]["update_error"] == "missing_prices"
assert all("/api/v3/" not in call[0] for call in session.calls)
jpath = next(p for p in (tmp_path / "o").iterdir() if p.suffix == ".json")
payload = json.loads(jpath.read_text(encoding="utf-8"))
assert payload["prices_source"] == "fmp"
assert payload["api_stats"] == {
"api_calls_made": 2,
"max_api_calls": 5,
"budget_remaining": 3,
"rate_limit_reached": False,
"failed_requests": 0,
}
assert payload["update"]["skipped_no_prices"] == 1
def test_main_update_partial_failure_saves_and_reports_count(tmp_path, monkeypatch, capsys):
book = tmp_path / "book.jsonl"
bmb.save_model_book(
book,
[bmb.make_record(cand(symbol="aaa")), bmb.make_record(cand(symbol="bbb"))],
)
session = FakeSession([FakeResponse(200, fmp_rows()), FakeResponse(500, None, "boom")])
install_fake_session(monkeypatch, session)
assert bmb.main(update_argv(book, tmp_path / "o")) == 0
capsys.readouterr()
recs = {r["symbol"]: r for r in bmb.load_model_book(book)}
assert recs["AAA"]["matured"] is True
assert recs["BBB"]["update_error"] == "missing_prices"
jpath = next(p for p in (tmp_path / "o").iterdir() if p.suffix == ".json")
payload = json.loads(jpath.read_text(encoding="utf-8"))
assert payload["api_stats"]["failed_requests"] == 1
def test_main_update_all_symbols_failed_is_fail_closed(tmp_path, monkeypatch, capsys):
book = tmp_path / "book.jsonl"
bmb.save_model_book(
book,
[bmb.make_record(cand(symbol="aaa")), bmb.make_record(cand(symbol="bbb"))],
)
before = book.read_bytes()
session = FakeSession([FakeResponse(500, None, "x"), FakeResponse(500, None, "y")])
install_fake_session(monkeypatch, session)
assert bmb.main(update_argv(book, tmp_path / "o")) == 1
err = capsys.readouterr().err
assert "ERROR: FMP price fetch failed for all 2 symbols; model book not updated" in err
assert book.read_bytes() == before
assert not (tmp_path / "o").exists() or not list((tmp_path / "o").iterdir())
def test_main_update_rate_limited_is_fail_closed(tmp_path, monkeypatch, capsys):
book = tmp_path / "book.jsonl"
bmb.save_model_book(book, [bmb.make_record(cand(symbol="aaa"))])
before = book.read_bytes()
session = FakeSession([FakeResponse(429), FakeResponse(429)])
install_fake_session(monkeypatch, session)
assert bmb.main(update_argv(book, tmp_path / "o")) == 1
assert "failed for all 1 symbols" in capsys.readouterr().err
assert book.read_bytes() == before
def test_main_update_all_empty_rows_still_saves(tmp_path, monkeypatch, capsys):
"""[] is data, not a failure: the no-rows ticker is marked missing_prices."""
book = tmp_path / "book.jsonl"
bmb.save_model_book(book, [bmb.make_record(cand(symbol="aaa"))])
install_fake_session(monkeypatch, FakeSession([FakeResponse(200, [])]))
assert bmb.main(update_argv(book, tmp_path / "o")) == 0
capsys.readouterr()
assert bmb.load_model_book(book)[0]["update_error"] == "missing_prices"
def test_main_update_budget_exhaustion_propagates(tmp_path, monkeypatch):
book = tmp_path / "book.jsonl"
bmb.save_model_book(book, [bmb.make_record(cand(symbol="aaa"))])
install_fake_session(monkeypatch, FakeSession([]))
with pytest.raises(bmb.ApiCallBudgetExceeded):
bmb.main(update_argv(book, tmp_path / "o", max_calls="0"))
def test_build_prices_for_update_skips_blank_symbols(monkeypatch):
session = FakeSession([])
install_fake_session(monkeypatch, session)
args = SimpleNamespace(prices_json=None, api_key=None, max_api_calls=3, history_days=10)
prices, stats = bmb.build_prices_for_update(args, [{"symbol": ""}, {}])
assert prices == {}
assert stats == {
"api_calls_made": 0,
"max_api_calls": 3,
"budget_remaining": 3,
"rate_limit_reached": False,
"failed_requests": 0,
}
assert session.calls == []
def test_main_summarize_end_to_end(tmp_path, capsys):
recs = []
for i in range(5):
r = bmb.make_record(cand(symbol=f"S{i}"))
r["matured"] = True
r["overall_outcome"] = "WORKED"
r["outcomes"] = {
"3d": {"forward_return_pct": 3.0},
"5d": {"forward_return_pct": 5.0, "mfe_pct": 8.0, "mae_pct": -1.0},
}
recs.append(r)
book = tmp_path / "book.jsonl"
bmb.save_model_book(book, recs)
out = tmp_path / "out"
rc = bmb.main(
[
"summarize",
"--model-book",
str(book),
"--group-by",
"rating, primary_trigger,",
"--min-sample",
"5",
"--output-dir",
str(out),
]
)
assert rc == 0
assert "Summarized 5 records" in capsys.readouterr().out
jpath = next(p for p in out.iterdir() if p.suffix == ".json")
payload = json.loads(jpath.read_text(encoding="utf-8"))
assert list(payload["groups"]) == ["rating", "primary_trigger"]
assert payload["model_book"] == str(book)
assert {r["group_by"] for r in payload["rule_candidates"]} == {"rating", "primary_trigger"}
md = next(p for p in out.iterdir() if p.suffix == ".md").read_text(encoding="utf-8")
assert "## Cohorts by rating" in md
assert "## Rule Candidates" in md
def test_main_requires_subcommand():
with pytest.raises(SystemExit) as exc:
bmb.main([])
assert exc.value.code == 2
def test_default_model_book_and_parser_defaults():
args = bmb.build_parser().parse_args(["summarize"])
assert args.model_book == bmb.DEFAULT_MODEL_BOOK
assert args.group_by == "rating,primary_trigger,setup_tags"
assert args.min_sample == 3
args = bmb.build_parser().parse_args(["update"])
assert args.only_pending is True
assert args.horizons == "3,5"
assert args.max_api_calls == 200
assert args.history_days == 80
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"])
SKILL.md
---
name: stockbee-setup-fluency-trainer
description: Build a Stockbee-style setup model book from momentum-burst screener candidates, then update 3-day and 5-day forward outcomes with MFE/MAE, stop-hit status, outcome tags, and cohort statistics. Use when the user wants to study Stockbee Momentum Burst examples, track failed candidates, build setup fluency, review A/B setup quality, or convert screener outputs into a learning loop rather than immediate trade signals.
---
# Stockbee Setup Fluency Trainer
Build and maintain a model book for Stockbee-style Momentum Burst setups. This skill turns daily screener candidates into structured study records, updates them after the 3-day and 5-day windows mature, and summarizes which setup features are working or failing.
## When to Use
- User wants to study Stockbee Momentum Burst setups systematically
- User asks to build a model book from `stockbee-momentum-burst-screener` output
- User wants to review failed candidates, missed trades, or A/B setup quality
- User wants 3-day / 5-day forward returns, MFE, MAE, and stop-hit outcomes
- User wants to improve setup recognition before increasing position size
- User asks which Stockbee tags should be promoted, downgraded, or filtered
## Prerequisites
- Python 3.10+
- A `stockbee-momentum-burst-screener` JSON report, or compatible candidate JSON
- Optional: FMP API key for outcome updates when offline OHLCV JSON is not supplied
- Recommended local state path: `state/stockbee/model_book.jsonl`
## Workflow
### Step 1: Ingest Momentum Burst Candidates
Run after the Stockbee Momentum Burst screener has produced a JSON report.
```bash
python3 skills/stockbee-setup-fluency-trainer/scripts/build_model_book.py ingest \
--screener-json reports/stockbee_momentum_burst_YYYY-MM-DD_HHMMSS.json \
--model-book state/stockbee/model_book.jsonl \
--output-dir reports/
```
Use `--include-rejects` when intentionally building a negative-example set. Otherwise rejected candidates are skipped.
### Step 2: Update 3-Day and 5-Day Outcomes
Use FMP:
```bash
python3 skills/stockbee-setup-fluency-trainer/scripts/build_model_book.py update \
--model-book state/stockbee/model_book.jsonl \
--horizons 3,5 \
--output-dir reports/
```
Use offline OHLCV JSON:
```bash
python3 skills/stockbee-setup-fluency-trainer/scripts/build_model_book.py update \
--model-book state/stockbee/model_book.jsonl \
--prices-json data/daily_ohlcv.json \
--horizons 3,5 \
--output-dir reports/
```
The update step records:
- Forward close return for each horizon
- MFE and MAE over each horizon
- Stop-hit status and first stop-hit date
- Outcome tags such as `STRONG_WINNER`, `WORKED`, `FAILED_STOP`, `FAILED_FADE`, `CHOPPY_FAILURE`, or `NEUTRAL`
### Step 3: Summarize Cohorts
```bash
python3 skills/stockbee-setup-fluency-trainer/scripts/build_model_book.py summarize \
--model-book state/stockbee/model_book.jsonl \
--group-by rating,primary_trigger,setup_tags \
--min-sample 5 \
--output-dir reports/
```
Review the generated Markdown and JSON reports. Treat `rule_candidates` as evidence prompts, not automatic rule changes.
### Step 4: Convert Evidence Into Practice
For cohorts with enough examples:
- Promote tags with high win rate, positive 5-day expectancy, and acceptable average MAE
- Downgrade or filter tags with weak 5-day expectancy, frequent stop hits, or repeated fade failures
- Inspect representative charts manually before changing trade rules
- Log accepted lessons in `trader-memory-core` or the monthly review process
## Model Book Fields
Each JSONL record includes:
- `record_id`, `symbol`, `setup_date`, `primary_trigger`
- `rating`, `setup_score`, `setup_tags`
- `entry_reference`, `stop_reference`, `risk_pct_to_stop`
- `human_label`, `human_decision`, `human_notes`
- `outcomes.3d` and `outcomes.5d`
- `overall_outcome`, `matured`, `raw_candidate`
## Interpretation Rules
- `STRONG_WINNER`: 5-day close return >= 8% or MFE >= 12%, with no stop hit
- `WORKED`: 5-day close return >= 4% or MFE >= 6%, with no stop hit
- `FAILED_STOP`: Stop was touched within the horizon
- `FAILED_FADE`: Forward return <= -2% without a recorded stop hit
- `CHOPPY_FAILURE`: Adverse excursion was large and forward progress was poor
- `NEUTRAL`: No decisive follow-through or failure
- `PENDING`: Not enough future bars yet
## Output
- `state/stockbee/model_book.jsonl` - Durable setup model book
- `stockbee_setup_fluency_ingest_YYYY-MM-DD_HHMMSS.json/md`
- `stockbee_setup_fluency_update_YYYY-MM-DD_HHMMSS.json/md`
- `stockbee_setup_fluency_summary_YYYY-MM-DD_HHMMSS.json/md`
## Resources
- `references/model_book_schema.md` - JSONL schema and lifecycle states
- `references/outcome_tags.md` - Outcome classification and tag definitions
- `references/review_workflow.md` - Daily, 3-day, 5-day, and monthly review routine
## Shared FMP client
Live FMP access uses the bundled generated `scripts/fmp_client.py`, loaded only
when an FMP run starts, so offline modes still work without `requests`. CLI
key/environment precedence, the API-call budget and successful output shapes are
preserved. Requests use stable endpoints only; a denied, empty or failed stable
request does not fall back to retired v3 URLs. The shared client masks raw and
URL-encoded key echoes, retries malformed JSON once, and retries a 429 at most once,
after a 60-second wait, before stopping on a repeated rate limit.
If every requested symbol fails at the transport level, `update` exits 1 and
leaves the model book unchanged instead of stamping each record `missing_prices`.