assets/bias_checklist.yaml
schema_version: "1.0"
items:
- id: look_ahead
title: Look-ahead bias
required: true
coverage_hint: point[- ]in[- ]time|look[- ]ahead|as.of|future data|no.future
- id: survivorship
title: Survivorship bias
required: true
coverage_hint: survivorship|delisted|non[- ]surviving|removed from the universe
- id: universe_selection
title: Universe selection
required: true
coverage_hint: universe|selection criteria|constituents|membership|screened .* pool|index membership
- id: delisted_securities
title: Delisted securities
required: true
coverage_hint: delist|trading suspended|halted|expired|formerly listed|removed from listing
- id: earnings_announcement_timing
title: Earnings announcement timing
required: true
coverage_hint: earnings.{0,20}(announce|report|timestamp|date|timing)|announcement timing|earnings date
- id: split_dividend_adjustment
title: Split and dividend adjustment
required: true
coverage_hint: split|dividend|adjusted|corporate action|restated|restatement|back[- ]adjusted
- id: transaction_costs_slippage
title: Transaction costs and slippage
required: true
coverage_hint: transaction cost|commission|slippage|fee[s]?|spread|after costs|net of costs
- id: short_borrow_availability
title: Short borrow availability
required: true
coverage_hint: short borrow|borrow fee|locate|hard.to borrow|HTB|short availability|rebate|shortable
- id: liquidity_capacity
title: Liquidity and capacity
required: true
coverage_hint: liquidity|capacity|ADV|average daily volume|volume cap|position cap|market impact
- id: parameter_multiplicity
title: Parameter multiplicity
required: true
coverage_hint: parameter.{0,15}(count|grid|combination|multiplicity|sweep)|multiple (parameters|configs)|overfit
- id: walk_forward_out_of_sample
title: Walk-forward and out-of-sample validation
required: true
coverage_hint: walk[- ]forward|out[- ]of[- ]sample|\bOOS\b|hold[- ]out|train[^ ]*test|deferred
- id: benchmark_attribution
title: Benchmark attribution
required: true
coverage_hint: benchmark|attribution|relative to (the )?(index|benchmark)|vs (index|benchmark)|beta[- ]adjusted
- id: regime_dependence
title: Regime dependence
required: true
coverage_hint: regime|sub[ ]?period|across (regime|market)|bull|bear|risk[- ]on|risk[- ]off|stress
references/overfitting_checklist.md
# Overfitting Detection Checklist
Heuristics for identifying strategy drafts that are likely overfit to historical data.
## Red Flags
### 1. Excessive Condition Count
Strategies with many entry conditions have fewer degrees of freedom and are more likely to describe noise rather than signal.
- **10+ total conditions** (entry + trend filter): Warning threshold
- **12+ total conditions**: Almost certainly overfit
### 2. Precise Threshold Values
Thresholds specified to decimal precision (e.g., "RSI > 33.5" vs "RSI > 30") suggest curve-fitting to historical data rather than capturing a robust pattern.
Detection: look for numbers containing a decimal point in condition strings.
Examples of precise thresholds:
- "RSI > 33.5" (precise: has decimal)
- "volume > 1.73 * avg" (precise: has decimal)
- "close > ma50 * 1.025" (precise: has decimal)
Examples of acceptable thresholds:
- "RSI > 30" (round number)
- "rel_volume >= 1.5" (half-step, commonly used)
- "close > ma50" (no threshold number with decimal)
### 3. Narrow Regime Specificity
Strategies designed for a single market regime (e.g., RiskOn only) with no validation across other regimes may not generalize.
### 4. Low Estimated Sample Size
If a strategy's conditions are so restrictive that fewer than 10 opportunities per year are expected, the backtest results are statistically unreliable.
### 5. Asymmetric Exit Parameters
- Stop losses wider than 15% suggest the strategy tolerates extreme adverse moves, which often indicates poor entry timing or overfit entry conditions.
- Risk-reward ratios below 1.5:1 require unrealistically high win rates to be profitable.
## Mitigation Strategies
1. Reduce condition count to essential filters only
2. Use round-number thresholds that capture behavioral levels (e.g., RSI 30/70, 50-day MA)
3. Validate across multiple regimes and time periods
4. Ensure sufficient sample size (30+ opportunities per year minimum)
5. Keep stop losses under 10% and target 2:1+ reward-to-risk
references/review_criteria.md
# Strategy Draft Review Criteria
Eight criteria (C1-C8) are applied to each strategy draft. Each criterion produces a score (0-100), a severity level, and optional revision instructions.
## Severity Levels
- **pass**: Score >= 60, no issues detected
- **warn**: Score 30-59, minor concerns that should be addressed
- **fail**: Score < 30, critical issues that must be resolved
## Verdict Logic
1. If C1 or C2 severity is "fail" -> immediate REJECT
2. Compute weighted confidence_score across all criteria
3. confidence_score >= 70 and no "fail" findings -> PASS
4. confidence_score < 35 -> REJECT
5. Otherwise -> REVISE (with revision instructions)
## C1: Edge Plausibility (Weight: 20)
Evaluate whether the strategy has a coherent, testable edge hypothesis.
| Condition | Severity | Score |
|-----------|----------|-------|
| thesis is empty or fewer than 5 words | fail | 10 |
| thesis is generic (no causal mechanism described) | warn | 40 |
| thesis contains specific causal reasoning | pass | 80 |
Generic thesis indicators: fewer than 10 words with no domain-specific terms like "momentum", "reversion", "drift", "earnings", "breakout", "gap", "volume", "sentiment".
## C2: Overfitting Risk (Weight: 20)
Assess the complexity of entry conditions relative to practical sample sizes.
| Condition | Severity | Score |
|-----------|----------|-------|
| conditions + trend_filter total > 12 | fail | 10 |
| conditions + trend_filter total > 10 | warn | 40 |
| conditions + trend_filter total <= 10 | pass | 80 |
Additional penalty: -10 per precise threshold detected in conditions (numbers with decimal points like "RSI > 33.5" or "volume > 1.73").
Minimum score after penalties: 0.
## C3: Sample Adequacy (Weight: 15)
Estimate annual trading opportunities and flag strategies that are too restrictive.
Estimation formula:
```
base = 252 (trading days)
if sector filter in conditions: base //= 3
if regime is not Neutral/Unknown/empty: base //= 2
base *= 0.8 ^ len(conditions)
base *= 0.85 ^ len(trend_filter)
result = max(base, 1)
```
| Estimated Opportunities | Severity | Score |
|------------------------|----------|-------|
| < 10 per year | fail | 10 |
| < 30 per year | warn | 40 |
| >= 30 per year | pass | 80 |
## C4: Regime Dependency (Weight: 10)
Check whether the strategy accounts for varying market regimes.
| Condition | Severity | Score |
|-----------|----------|-------|
| Single regime with no cross-regime validation plan | warn | 40 |
| Otherwise | pass | 80 |
A strategy has cross-regime validation if its validation_plan mentions "regime" or "regimes" (case-insensitive) in any of its values.
## C5: Exit Calibration (Weight: 10)
Validate stop-loss and take-profit parameters.
| Condition | Severity | Score |
|-----------|----------|-------|
| stop_loss_pct > 0.15 | fail | 10 |
| take_profit_rr < 1.5 | fail | 10 |
| Both within range | pass | 80 |
If both fail conditions are met, score is 10 (minimum).
## C6: Risk Concentration (Weight: 10)
Evaluate position sizing and concentration limits.
| Condition | Severity | Score |
|-----------|----------|-------|
| risk_per_trade > 0.02 | fail | 10 |
| risk_per_trade > 0.015 | warn | 40 |
| max_positions > 10 | fail | 10 |
| All within range | pass | 80 |
If multiple fail conditions met, use the worst (lowest) score.
## C7: Execution Realism (Weight: 10)
Check for practical execution concerns.
| Condition | Severity | Score |
|-----------|----------|-------|
| No volume filter in conditions | warn | 50 |
| export_ready_v1=true but entry_family not in EXPORTABLE_FAMILIES | fail | 10 |
| Otherwise | pass | 80 |
EXPORTABLE_FAMILIES = {"pivot_breakout", "gap_up_continuation"}
If both conditions trigger, use the worst (lowest) score and severity.
## C8: Invalidation Quality (Weight: 5)
Assess the quality of invalidation signals.
| Condition | Severity | Score |
|-----------|----------|-------|
| invalidation_signals is empty | fail | 10 |
| invalidation_signals has fewer than 2 entries | warn | 40 |
| 2 or more entries | pass | 80 |
## Weight Summary
| Criterion | Weight |
|-----------|--------|
| C1 Edge Plausibility | 20 |
| C2 Overfitting Risk | 20 |
| C3 Sample Adequacy | 15 |
| C4 Regime Dependency | 10 |
| C5 Exit Calibration | 10 |
| C6 Risk Concentration | 10 |
| C7 Execution Realism | 10 |
| C8 Invalidation Quality | 5 |
| **Total** | **100** |
requirements.txt
# Runtime dependencies for standalone .skill installs (issue #330).
# Verified by scripts/check_skill_deps.py.
PyYAML>=6.0
scripts/review_strategy_drafts.py
#!/usr/bin/env python3
"""Critically review strategy drafts for edge plausibility, overfitting risk,
sample size adequacy, and execution realism.
Outputs PASS / REVISE / REJECT verdicts with confidence scores.
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import yaml
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
DEFAULT_EXPORTABLE_FAMILIES = {
"pivot_breakout",
"gap_up_continuation",
"panic_reversal",
"news_reaction",
}
WEIGHTS: dict[str, int] = {
"C1_edge_plausibility": 20,
"C2_overfitting_risk": 20,
"C3_sample_adequacy": 15,
"C4_regime_dependency": 10,
"C5_exit_calibration": 10,
"C6_risk_concentration": 10,
"C7_execution_realism": 10,
"C8_invalidation_quality": 5,
}
DOMAIN_TERMS = {
"momentum",
"reversion",
"drift",
"earnings",
"breakout",
"gap",
"volume",
"sentiment",
"mean",
"reversal",
"trend",
"liquidity",
"participation",
"institutional",
"continuation",
"contraction",
"expansion",
"pivot",
"vcp",
"base",
}
MECHANISM_KEYWORDS = {
"participation",
"drift",
"overreaction",
"accumulation",
"exhaustion",
"imbalance",
"herding",
"continuation",
"underreaction",
"institutional",
}
_PRECISE_THRESHOLD_RE = re.compile(r"(?:\b\d{2,}\.\d+\b|\b\d+\.\d{2,}\b)")
class ReviewError(Exception):
"""Raised when review cannot proceed."""
# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class ReviewFinding:
criterion: str
score: int
severity: str # "pass" / "warn" / "fail"
reason: str
revision_instruction: str | None = None
@dataclass
class DraftReview:
draft_id: str
verdict: str # PASS / REVISE / REJECT
confidence_score: int
export_eligible: bool
findings: list[ReviewFinding] = field(default_factory=list)
revision_instructions: list[str] = field(default_factory=list)
# ---------------------------------------------------------------------------
# Criterion evaluators (C1 – C8)
# ---------------------------------------------------------------------------
def _severity_from_score(score: int) -> str:
if score < 30:
return "fail"
if score < 60:
return "warn"
return "pass"
def evaluate_c1(draft: dict) -> ReviewFinding:
"""C1: Edge Plausibility."""
thesis = str(draft.get("thesis", "")).strip()
words = thesis.split()
if not thesis or len(words) < 3:
return ReviewFinding(
criterion="C1_edge_plausibility",
score=10,
severity="fail",
reason=f"Thesis too short ({len(words)} words). Must articulate a causal edge hypothesis.",
revision_instruction="Expand thesis to describe the causal mechanism behind the expected edge.",
)
lower = thesis.lower()
has_domain = any(term in lower for term in DOMAIN_TERMS)
if len(words) < 10 and not has_domain:
return ReviewFinding(
criterion="C1_edge_plausibility",
score=40,
severity="warn",
reason="Thesis is generic with no domain-specific causal reasoning.",
revision_instruction="Add specific causal mechanism (e.g., institutional participation, earnings drift).",
)
# Continuous scoring for the pass path
domain_count = sum(1 for term in DOMAIN_TERMS if term in lower)
score = 50 + min(domain_count * 10, 30)
if len(words) >= 10:
score += 10
if len(words) > 20:
score += 10
has_mechanism = any(kw in lower for kw in MECHANISM_KEYWORDS)
if has_mechanism:
score += 10
score = min(score, 95)
return ReviewFinding(
criterion="C1_edge_plausibility",
score=score,
severity="pass",
reason="Thesis contains specific causal reasoning.",
)
def evaluate_c2(draft: dict) -> ReviewFinding:
"""C2: Overfitting Risk."""
entry = draft.get("entry", {})
conditions = entry.get("conditions", []) if isinstance(entry.get("conditions"), list) else []
trend_filter = (
entry.get("trend_filter", []) if isinstance(entry.get("trend_filter"), list) else []
)
total = len(conditions) + len(trend_filter)
if total >= 13:
score = 10
elif total >= 11:
score = 40
elif total >= 8:
score = 60
elif total >= 5:
score = 80
else:
score = 90
# Precise threshold penalty
precise_count = 0
for c in conditions:
if isinstance(c, str):
precise_count += len(_PRECISE_THRESHOLD_RE.findall(c))
score = max(score - 10 * precise_count, 0)
severity = _severity_from_score(score)
parts = []
if total > 10:
parts.append(
f"{total} combined filters ({len(conditions)} conditions + {len(trend_filter)} trend)"
)
if precise_count:
parts.append(f"{precise_count} precise threshold(s) detected")
reason = "; ".join(parts) if parts else f"{total} filters within acceptable range."
instruction = None
if severity != "pass":
instructions = []
if total > 10:
instructions.append("Reduce entry conditions to 5 most predictive")
if precise_count:
instructions.append("Round precise thresholds to standard increments")
instruction = "; ".join(instructions)
return ReviewFinding(
criterion="C2_overfitting_risk",
score=score,
severity=severity,
reason=reason,
revision_instruction=instruction,
)
def estimate_annual_opportunities(draft: dict) -> int:
"""Estimate yearly trading signals from condition restrictiveness."""
entry = draft.get("entry", {})
conditions = entry.get("conditions", []) if isinstance(entry.get("conditions"), list) else []
trend_filter = (
entry.get("trend_filter", []) if isinstance(entry.get("trend_filter"), list) else []
)
base = 252
if any("sector" in str(c).lower() for c in conditions):
base //= 3
regime = str(draft.get("regime", ""))
if regime not in ("", "Unknown", "Neutral"):
base //= 2
base = int(base * (0.8 ** len(conditions)))
base = int(base * (0.85 ** len(trend_filter)))
return max(base, 1)
def _c3_score_from_estimate(est: int) -> int:
"""Map estimated annual opportunities to a continuous score."""
if est < 10:
return 10
if est <= 29:
# Linear interpolation 30..59 over est 10..29
return 30 + int(round((est - 10) / (29 - 10) * (59 - 30)))
if est <= 49:
# Linear interpolation 65..74 over est 30..49
return 65 + int(round((est - 30) / (49 - 30) * (74 - 65)))
if est <= 99:
# Linear interpolation 75..85 over est 50..99
return 75 + int(round((est - 50) / (99 - 50) * (85 - 75)))
# est >= 100: linear interpolation 85..95 capped at 95 for est>=200
capped = min(est, 200)
return 85 + int(round((capped - 100) / (200 - 100) * (95 - 85)))
def evaluate_c3(draft: dict) -> ReviewFinding:
"""C3: Sample Adequacy."""
est = estimate_annual_opportunities(draft)
score = _c3_score_from_estimate(est)
severity = _severity_from_score(score)
reason = f"Estimated {est} annual opportunities."
instruction = None
if severity != "pass":
instruction = (
"Relax conditions or remove sector/regime restrictions to increase sample size."
)
return ReviewFinding(
criterion="C3_sample_adequacy",
score=score,
severity=severity,
reason=reason,
revision_instruction=instruction,
)
def evaluate_c4(draft: dict) -> ReviewFinding:
"""C4: Regime Dependency."""
regime = str(draft.get("regime", ""))
is_single = regime not in ("", "Unknown", "Neutral")
if not is_single:
return ReviewFinding(
criterion="C4_regime_dependency",
score=80,
severity="pass",
reason="Strategy not restricted to single regime.",
)
# Check for cross-regime validation
vp = draft.get("validation_plan", {})
if isinstance(vp, dict):
vp_text = json.dumps(vp).lower()
if "regime" in vp_text:
return ReviewFinding(
criterion="C4_regime_dependency",
score=80,
severity="pass",
reason=f"Single regime ({regime}) but cross-regime validation planned.",
)
return ReviewFinding(
criterion="C4_regime_dependency",
score=40,
severity="warn",
reason=f"Restricted to {regime} with no cross-regime validation plan.",
revision_instruction="Add cross-regime validation to success criteria.",
)
def evaluate_c5(draft: dict) -> ReviewFinding:
"""C5: Exit Calibration."""
exit_conf = draft.get("exit", {})
stop = float(exit_conf.get("stop_loss_pct", 0))
rr = float(exit_conf.get("take_profit_rr", 0))
problems = []
if stop > 0.15:
problems.append(f"stop_loss_pct={stop:.0%} exceeds 15%")
if rr < 1.5:
problems.append(f"take_profit_rr={rr:.1f} below 1.5")
if problems:
score = 10
severity = "fail"
reason = "; ".join(problems)
instruction = "Tighten stop-loss to <=15% and ensure reward-to-risk >= 1.5."
else:
score, severity = 80, "pass"
reason = f"Exit parameters acceptable (stop={stop:.0%}, RR={rr:.1f})."
instruction = None
return ReviewFinding(
criterion="C5_exit_calibration",
score=score,
severity=severity,
reason=reason,
revision_instruction=instruction,
)
def evaluate_c6(draft: dict) -> ReviewFinding:
"""C6: Risk Concentration."""
risk = draft.get("risk", {})
rpt = float(risk.get("risk_per_trade", 0))
mp = int(risk.get("max_positions", 0))
problems = []
worst_score = 80
if rpt > 0.02:
problems.append(f"risk_per_trade={rpt:.1%} exceeds 2%")
worst_score = min(worst_score, 10)
elif rpt > 0.015:
problems.append(f"risk_per_trade={rpt:.1%} exceeds 1.5%")
worst_score = min(worst_score, 40)
if mp > 10:
problems.append(f"max_positions={mp} exceeds 10")
worst_score = min(worst_score, 10)
severity = _severity_from_score(worst_score)
reason = "; ".join(problems) if problems else "Risk parameters within acceptable range."
instruction = None
if severity != "pass":
instruction = "Reduce risk_per_trade to <=1.5% and max_positions to <=10."
return ReviewFinding(
criterion="C6_risk_concentration",
score=worst_score,
severity=severity,
reason=reason,
revision_instruction=instruction,
)
def evaluate_c7(draft: dict, exportable_families: set[str] | None = None) -> ReviewFinding:
"""C7: Execution Realism."""
families = (
exportable_families if exportable_families is not None else DEFAULT_EXPORTABLE_FAMILIES
)
entry = draft.get("entry", {})
conditions = entry.get("conditions", []) if isinstance(entry.get("conditions"), list) else []
has_volume = any("volume" in str(c).lower() for c in conditions)
export_ready = bool(draft.get("export_ready_v1", False))
family = str(draft.get("entry_family", ""))
bad_export = export_ready and family not in families
if bad_export:
return ReviewFinding(
criterion="C7_execution_realism",
score=10,
severity="fail",
reason=f"export_ready_v1=true but entry_family='{family}' not in exportable families.",
revision_instruction=f"Set export_ready_v1=false or change entry_family to one of {sorted(families)}.",
)
if not has_volume:
return ReviewFinding(
criterion="C7_execution_realism",
score=50,
severity="warn",
reason="No volume filter in entry conditions.",
revision_instruction="Add volume filter (e.g., avg_volume > 500000) for execution realism.",
)
return ReviewFinding(
criterion="C7_execution_realism",
score=80,
severity="pass",
reason="Volume filter present and export settings consistent.",
)
def evaluate_c8(draft: dict) -> ReviewFinding:
"""C8: Invalidation Quality."""
signals = draft.get("invalidation_signals", [])
if not isinstance(signals, list):
signals = []
if len(signals) == 0:
score, severity = 10, "fail"
reason = "No invalidation signals defined."
instruction = "Add at least 2 concrete invalidation signals."
elif len(signals) < 2:
score, severity = 40, "warn"
reason = f"Only {len(signals)} invalidation signal(s); minimum 2 recommended."
instruction = "Add additional invalidation signal for robustness."
else:
score, severity = 80, "pass"
reason = f"{len(signals)} invalidation signals defined."
instruction = None
return ReviewFinding(
criterion="C8_invalidation_quality",
score=score,
severity=severity,
reason=reason,
revision_instruction=instruction,
)
# ---------------------------------------------------------------------------
# Review engine
# ---------------------------------------------------------------------------
ALL_EVALUATORS = [
evaluate_c1,
evaluate_c2,
evaluate_c3,
evaluate_c4,
evaluate_c5,
evaluate_c6,
evaluate_c7,
evaluate_c8,
]
def compute_confidence_score(findings: list[ReviewFinding]) -> int:
"""Weighted average of finding scores."""
total_weight = 0
weighted_sum = 0
for f in findings:
w = WEIGHTS.get(f.criterion, 0)
weighted_sum += f.score * w
total_weight += w
if total_weight == 0:
return 0
return int(round(weighted_sum / total_weight))
def determine_verdict(findings: list[ReviewFinding], confidence: int) -> str:
"""Determine PASS / REVISE / REJECT."""
# C1 or C2 fail → immediate REJECT
for f in findings:
if f.criterion in ("C1_edge_plausibility", "C2_overfitting_risk") and f.severity == "fail":
return "REJECT"
has_fail = any(f.severity == "fail" for f in findings)
if confidence >= 70 and not has_fail:
return "PASS"
if confidence < 35:
return "REJECT"
return "REVISE"
def is_export_eligible(
draft: dict, verdict: str, exportable_families: set[str] | None = None
) -> bool:
"""Check if draft qualifies for pipeline export."""
families = (
exportable_families if exportable_families is not None else DEFAULT_EXPORTABLE_FAMILIES
)
return (
verdict == "PASS"
and bool(draft.get("export_ready_v1", False))
and str(draft.get("entry_family", "")) in families
)
BUNDLED_CHECKLIST = Path(__file__).resolve().parent.parent / "assets" / "bias_checklist.yaml"
BIAS_SCHEMA_VERSION = "1.0"
def load_bias_checklist(path: Path) -> list[dict]:
"""Load and validate bias checklist items from a YAML file.
The checklist schema is ``{"schema_version": str, "items": [{id, title,
required, coverage_hint}]}``. Items missing ``id`` or ``coverage_hint`` are
rejected so the reviewer never silently skips a required rule.
"""
try:
data = yaml.safe_load(path.read_text(encoding="utf-8"))
except (OSError, yaml.YAMLError) as exc:
raise ReviewError(f"Could not read bias checklist {path}: {exc}") from exc
if not isinstance(data, dict) or not isinstance(data.get("items"), list):
raise ReviewError(f"Bias checklist {path} must contain an 'items' list")
declared = data.get("schema_version")
if declared is not None and str(declared) != BIAS_SCHEMA_VERSION:
raise ReviewError(
f"Bias checklist {path} schema_version {declared!r} "
f"does not match expected {BIAS_SCHEMA_VERSION!r}"
)
items: list[dict] = []
for raw in data["items"]:
if not isinstance(raw, dict) or not raw.get("id") or not raw.get("coverage_hint"):
raise ReviewError(f"Invalid bias checklist item in {path}: {raw!r}")
items.append(
{
"id": str(raw["id"]),
"title": str(raw.get("title", raw["id"])),
"required": bool(raw.get("required", True)),
"coverage_hint": str(raw["coverage_hint"]),
}
)
return items
def _draft_text(draft: dict) -> str:
"""Flatten all scalar text in a draft (excluding its bias block) for hint matching."""
parts: list[str] = []
def walk(obj: Any) -> None:
if isinstance(obj, dict):
for key, value in obj.items():
if key == "bias_checklist":
continue
walk(value)
elif isinstance(obj, list):
for value in obj:
walk(value)
elif isinstance(obj, (str, int, float, bool)):
parts.append(str(obj))
walk(draft)
return " ".join(parts)
def evaluate_bias_checklist(draft: dict, items: list[dict]) -> list[dict]:
"""Score one draft against the bias checklist.
An item is ``addressed`` iff the draft declares it in its ``bias_checklist``
block with a truthy value (a dict with ``addressed`` truthy and defaulting to
true when absent; a plain truthy scalar is also addressed). ``noted_in_draft``
is an informational heuristic that reports whether the item's concept text
appears anywhere in the draft, so an author can see that a rule is discussed
but not yet declared. Escalation in :func:`apply_bias_gate` is driven solely
by ``addressed``.
"""
declared = draft.get("bias_checklist")
declared_map = declared if isinstance(declared, dict) else {}
text = _draft_text(draft)
results: list[dict] = []
for item in items:
value = declared_map.get(item["id"])
if isinstance(value, dict):
addressed = bool(value.get("addressed", True))
note = str(value.get("note", ""))
else:
addressed = bool(value)
note = str(value) if value else ""
try:
notion = re.search(item["coverage_hint"], text, re.IGNORECASE)
except re.error:
notion = None
results.append(
{
"item_id": item["id"],
"title": item["title"],
"required": item["required"],
"addressed": addressed,
"noted_in_draft": bool(notion),
"note": note,
}
)
return results
def apply_bias_gate(review: DraftReview, bias_results: list[dict]) -> bool:
"""Downgrade PASS -> REVISE when a required bias item is unaddressed.
Returns True when the verdict was downgraded. This is non-scoring: it never
changes the confidence score, C1-C8 weights, or a REJECT verdict. On
downgrade, export eligibility is cleared to match the ``strict_export``
behavior, so a REVISE draft is never export-eligible.
"""
unaddressed = [r for r in bias_results if r["required"] and not r["addressed"]]
if not unaddressed:
return False
if review.verdict == "PASS":
review.verdict = "REVISE"
review.export_eligible = False
names = ", ".join(r["item_id"] for r in unaddressed)
review.revision_instructions.append(
f"Declare how {names} {('is' if len(unaddressed) == 1 else 'are')} "
"addressed in the draft's bias_checklist block"
)
return True
return False
def _review_bias_for_all(
drafts: list[dict], reviews: list[DraftReview], bias_items: list[dict]
) -> dict[str, list[dict]]:
"""Evaluate the bias checklist for every draft and apply the verdict gate.
Drafts and reviews are joined positionally (``review_draft`` preserves the
``load_drafts_from_dir`` ordering), so the ``bias_by_id`` map is keyed by
each review's ``draft_id``.
"""
bias_by_id: dict[str, list[dict]] = {}
for draft, review in zip(drafts, reviews):
results = evaluate_bias_checklist(draft, bias_items)
apply_bias_gate(review, results)
bias_by_id[review.draft_id] = results
return bias_by_id
def review_draft(
draft: dict, *, strict_export: bool = False, exportable_families: set[str] | None = None
) -> DraftReview:
"""Review a single strategy draft.
When *strict_export* is True, export-eligible drafts that have any
warn-level findings are downgraded from PASS to REVISE so that the
warnings must be resolved before export.
"""
draft_id = str(draft.get("id", "unknown"))
findings = [
ev(draft, exportable_families) if ev is evaluate_c7 else ev(draft) for ev in ALL_EVALUATORS
]
confidence = compute_confidence_score(findings)
verdict = determine_verdict(findings, confidence)
export_ok = is_export_eligible(draft, verdict, exportable_families)
# Strict export: export-eligible PASS with any warn → REVISE
if strict_export and verdict == "PASS" and export_ok:
has_warn = any(f.severity == "warn" for f in findings)
if has_warn:
verdict = "REVISE"
export_ok = False
instructions: list[str] = []
if verdict == "REVISE":
instructions = [
f.revision_instruction
for f in findings
if f.revision_instruction and f.severity in ("warn", "fail")
]
return DraftReview(
draft_id=draft_id,
verdict=verdict,
confidence_score=confidence,
export_eligible=export_ok,
findings=findings,
revision_instructions=instructions,
)
# ---------------------------------------------------------------------------
# I/O helpers
# ---------------------------------------------------------------------------
def load_draft_file(path: Path) -> dict:
"""Load a single draft YAML."""
text = path.read_text()
data = yaml.safe_load(text)
if not isinstance(data, dict):
raise ReviewError(f"Invalid draft format in {path}: expected mapping")
return data
def load_drafts_from_dir(drafts_dir: Path) -> list[dict]:
"""Load all YAML drafts from a directory."""
files = sorted(drafts_dir.glob("*.yaml"))
if not files:
raise ReviewError(f"No .yaml files found in {drafts_dir}")
drafts = []
for f in files:
# Skip manifest or non-draft files
if f.name.startswith("run_manifest"):
continue
try:
drafts.append(load_draft_file(f))
except Exception as exc:
print(f"[WARN] Skipping {f.name}: {exc}", file=sys.stderr)
return drafts
def build_output(
drafts_source: str,
draft_count: int,
reviews: list[DraftReview],
bias_by_id: dict[str, list[dict]] | None = None,
) -> dict[str, Any]:
"""Build the output payload.
When ``bias_by_id`` is supplied, each review that has a matching key gets a
``bias_review`` block merged into its serialized dict. When it is not
supplied (checklist flag off) the key is omitted entirely so the output is
byte-identical to before the bias feature.
"""
summary = {"total": draft_count, "PASS": 0, "REVISE": 0, "REJECT": 0, "export_eligible": 0} # nosec B105
for r in reviews:
summary[r.verdict] = summary.get(r.verdict, 0) + 1
if r.export_eligible:
summary["export_eligible"] += 1
reviews_payload: list[dict[str, Any]] = []
for r in reviews:
payload = asdict(r)
if bias_by_id and r.draft_id in bias_by_id:
payload["bias_review"] = bias_by_id[r.draft_id]
reviews_payload.append(payload)
return {
"generated_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
"source": {
"drafts_dir": drafts_source,
"draft_count": draft_count,
},
"summary": summary,
"reviews": reviews_payload,
}
def build_markdown_summary(output: dict) -> str:
"""Generate a human-readable markdown summary."""
lines = [
"# Strategy Draft Review Summary",
"",
f"**Generated:** {output['generated_at_utc']}",
f"**Source:** {output['source']['drafts_dir']}",
f"**Drafts reviewed:** {output['source']['draft_count']}",
"",
"## Verdict Summary",
"",
"| Verdict | Count |",
"|---------|-------|",
]
s = output["summary"]
for v in ("PASS", "REVISE", "REJECT"):
lines.append(f"| {v} | {s.get(v, 0)} |")
lines.append(f"| Export Eligible | {s.get('export_eligible', 0)} |")
lines.append("")
lines.append("## Individual Reviews")
lines.append("")
for r in output.get("reviews", []):
verdict_icon = {"PASS": "PASS", "REVISE": "REVISE", "REJECT": "REJECT"}.get( # nosec B105
r["verdict"], r["verdict"]
)
lines.append(f"### {r['draft_id']} — {verdict_icon} (confidence: {r['confidence_score']})")
if r.get("export_eligible"):
lines.append("**Export eligible**")
lines.append("")
lines.append("| Criterion | Score | Severity | Reason |")
lines.append("|-----------|-------|----------|--------|")
for f in r.get("findings", []):
lines.append(f"| {f['criterion']} | {f['score']} | {f['severity']} | {f['reason']} |")
if r.get("bias_review"):
unaddressed = [b for b in r["bias_review"] if b["required"] and not b["addressed"]]
lines.append("")
lines.append("**Bias Checklist:**")
lines.append("")
lines.append("| Item | Required | Addressed | Note |")
lines.append("|------|----------|-----------|------|")
for b in r["bias_review"]:
lines.append(
f"| {b['item_id']} | {b['required']} | {b['addressed']} | {b.get('note', '')} |"
)
if unaddressed:
lines.append("")
lines.append(
"_Unaddressed required items: "
+ ", ".join(b["item_id"] for b in unaddressed)
+ "_"
)
if r.get("revision_instructions"):
lines.append("")
lines.append("**Revision Instructions:**")
for inst in r["revision_instructions"]:
lines.append(f"- {inst}")
lines.append("")
return "\n".join(lines) + "\n"
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"""Parse command-line arguments."""
parser = argparse.ArgumentParser(
description="Review strategy drafts for edge plausibility and quality.",
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--drafts-dir", help="Directory containing strategy draft YAML files")
group.add_argument("--draft", help="Path to a single strategy draft YAML file")
parser.add_argument(
"--output-dir", default="reports/", help="Output directory (default: reports/)"
)
parser.add_argument(
"--format",
dest="output_format",
choices=["yaml", "json"],
default="yaml",
help="Output format (default: yaml)",
)
parser.add_argument(
"--markdown-summary",
action="store_true",
help="Generate additional markdown summary file",
)
parser.add_argument(
"--strict-export",
action="store_true",
default=False,
help="Export-eligible drafts with any warn finding get REVISE instead of PASS",
)
parser.add_argument(
"--exportable-families",
default=None,
help="Comma-separated list of exportable entry families (overrides module default)",
)
parser.add_argument(
"--bias-checklist",
nargs="?",
const=BUNDLED_CHECKLIST,
default=None,
help=(
"Enable the strategy-research bias checklist review against the bundled "
"checklist (default) or a custom YAML path. Without this flag, the bias "
"review is not applied and output matches prior behavior."
),
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
"""CLI entrypoint."""
args = parse_args(argv)
output_dir = Path(args.output_dir).resolve()
ef_override: set[str] | None = None
if args.exportable_families:
ef_override = {f.strip() for f in args.exportable_families.split(",") if f.strip()}
try:
if args.drafts_dir:
drafts_dir = Path(args.drafts_dir).resolve()
if not drafts_dir.is_dir():
raise ReviewError(f"Drafts directory not found: {drafts_dir}")
drafts = load_drafts_from_dir(drafts_dir)
source = str(drafts_dir)
else:
draft_path = Path(args.draft).resolve()
if not draft_path.is_file():
raise ReviewError(f"Draft file not found: {draft_path}")
drafts = [load_draft_file(draft_path)]
source = str(draft_path)
if not drafts:
raise ReviewError("No valid drafts to review.")
reviews = [
review_draft(d, strict_export=args.strict_export, exportable_families=ef_override)
for d in drafts
]
# When --bias-checklist is supplied without a path, the bundled checklist
# under assets/ is used. Without the flag, bias is not applied and the
# output matches prior behavior.
bias_by_id: dict[str, list[dict]] | None = None
if args.bias_checklist is not None:
checklist_path = Path(args.bias_checklist).resolve()
if not checklist_path.is_file():
raise ReviewError(f"Bias checklist not found: {checklist_path}")
bias_items = load_bias_checklist(checklist_path)
bias_by_id = _review_bias_for_all(drafts, reviews, bias_items)
output = build_output(source, len(drafts), reviews, bias_by_id)
output_dir.mkdir(parents=True, exist_ok=True)
ext = args.output_format
out_path = output_dir / f"review.{ext}"
if ext == "yaml":
out_path.write_text(
yaml.safe_dump(output, sort_keys=False, default_flow_style=False), encoding="utf-8"
)
else:
out_path.write_text(json.dumps(output, indent=2) + "\n", encoding="utf-8")
if args.markdown_summary:
md_path = output_dir / "review_summary.md"
md_path.write_text(build_markdown_summary(output), encoding="utf-8")
except ReviewError as exc:
print(f"[ERROR] {exc}", file=sys.stderr)
return 1
s = output["summary"]
print(
f"[OK] Reviewed {s['total']} drafts: "
f"PASS={s['PASS']} REVISE={s['REVISE']} REJECT={s['REJECT']} "
f"export_eligible={s['export_eligible']} output={out_path}"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/tests/conftest.py
"""Test configuration for edge-strategy-reviewer scripts."""
from __future__ import annotations
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
@pytest.fixture()
def well_formed_breakout_draft() -> dict:
"""A well-formed breakout draft that should PASS all criteria."""
return {
"id": "draft_edge_concept_breakout_behavior_riskon_core",
"concept_id": "edge_concept_breakout_behavior_riskon",
"name": "Participation-backed trend breakout (core)",
"variant": "core",
"export_ready_v1": True,
"entry_family": "pivot_breakout",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"thesis": "Price breakouts above 20-day highs with above-average volume indicate institutional participation and momentum continuation.",
"entry": {
"conditions": [
"close > high20_prev",
"rel_volume >= 2",
"close > ma50 > ma200",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
},
"exit": {
"stop_loss_pct": 0.07,
"take_profit_rr": 3.0,
},
"risk": {
"risk_per_trade": 0.01,
"max_positions": 5,
"max_sector_exposure": 0.3,
},
"invalidation_signals": [
"Breakout fails within 3 days",
"Volume dries up below average",
"Broad market enters risk-off regime",
],
"validation_plan": {
"period": "2016-01-01 to latest",
"entry_timing": "next_open",
"hold_days": [5, 20, 60],
"success_criteria": [
"expected_value_after_costs > 0",
"stable across regimes and subperiods",
],
},
}
@pytest.fixture()
def empty_thesis_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with empty thesis -- should fail C1."""
d = dict(well_formed_breakout_draft)
d["thesis"] = ""
return d
@pytest.fixture()
def generic_thesis_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with generic thesis (short, no domain terms) -- should warn C1."""
d = dict(well_formed_breakout_draft)
d["thesis"] = "Buy stocks that are going up soon"
return d
@pytest.fixture()
def excessive_conditions_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with 11 conditions+trend_filter -- should warn C2."""
d = dict(well_formed_breakout_draft)
d["entry"] = {
"conditions": [
"close > high20_prev",
"rel_volume >= 2",
"close > ma50",
"close > ma200",
"RSI > 50",
"MACD > 0",
"ADX > 25",
"close > vwap",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
}
return d
@pytest.fixture()
def extreme_conditions_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with 13 conditions+trend_filter -- should fail C2."""
d = dict(well_formed_breakout_draft)
d["entry"] = {
"conditions": [
"close > high20_prev",
"rel_volume >= 2",
"close > ma50",
"close > ma200",
"RSI > 50",
"MACD > 0",
"ADX > 25",
"close > vwap",
"stochastic > 80",
"OBV rising",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
}
return d
@pytest.fixture()
def precise_threshold_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with precise decimal thresholds -- should get C2 penalty."""
d = dict(well_formed_breakout_draft)
d["entry"] = {
"conditions": [
"RSI > 33.5",
"volume > 1.73 * avg_volume",
"close > ma50 * 1.025",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
}
return d
@pytest.fixture()
def restrictive_conditions_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with regime + many conditions -- should warn C3 (est ~20/yr)."""
d = dict(well_formed_breakout_draft)
d["regime"] = "RiskOn"
d["entry"] = {
"conditions": [
"close > high20_prev",
"rel_volume >= 2",
"close > ma50",
"close > ma200",
"RSI > 50",
"MACD > 0",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
}
return d
@pytest.fixture()
def extreme_restriction_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with extreme restriction -- should fail C3."""
d = dict(well_formed_breakout_draft)
d["regime"] = "RiskOn"
d["entry"] = {
"conditions": [
"sector == Technology",
"close > high20_prev",
"rel_volume >= 1.5",
"close > ma50",
"close > ma200",
"RSI > 50",
"MACD > 0",
"ADX > 25",
"close > vwap",
"stochastic > 80",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
"sma_20 > sma_50",
],
}
return d
@pytest.fixture()
def single_regime_no_validation_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with single regime and no cross-regime validation -- should warn C4."""
d = dict(well_formed_breakout_draft)
d["regime"] = "RiskOn"
d["validation_plan"] = {
"period": "2016-01-01 to latest",
"entry_timing": "next_open",
"hold_days": [5, 20, 60],
"success_criteria": [
"expected_value_after_costs > 0",
],
}
return d
@pytest.fixture()
def unreasonable_stop_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with stop_loss > 15% -- should fail C5."""
d = dict(well_formed_breakout_draft)
d["exit"] = {
"stop_loss_pct": 0.20,
"take_profit_rr": 3.0,
}
return d
@pytest.fixture()
def low_rr_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with take_profit_rr < 1.5 -- should fail C5."""
d = dict(well_formed_breakout_draft)
d["exit"] = {
"stop_loss_pct": 0.07,
"take_profit_rr": 1.0,
}
return d
@pytest.fixture()
def high_risk_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with risk_per_trade > 1.5% -- should warn C6."""
d = dict(well_formed_breakout_draft)
d["risk"] = {
"risk_per_trade": 0.018,
"max_positions": 5,
"max_sector_exposure": 0.3,
}
return d
@pytest.fixture()
def extreme_risk_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with risk_per_trade > 2% -- should fail C6."""
d = dict(well_formed_breakout_draft)
d["risk"] = {
"risk_per_trade": 0.025,
"max_positions": 5,
"max_sector_exposure": 0.3,
}
return d
@pytest.fixture()
def no_volume_filter_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with no volume filter in conditions -- should warn C7."""
d = dict(well_formed_breakout_draft)
d["entry"] = {
"conditions": [
"close > high20_prev",
"close > ma50 > ma200",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
}
return d
@pytest.fixture()
def wrong_family_export_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with export_ready_v1=true but non-exportable family -- should fail C7."""
d = dict(well_formed_breakout_draft)
d["export_ready_v1"] = True
d["entry_family"] = "mean_reversion"
return d
@pytest.fixture()
def empty_invalidation_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with empty invalidation_signals -- should fail C8."""
d = dict(well_formed_breakout_draft)
d["invalidation_signals"] = []
return d
@pytest.fixture()
def insufficient_invalidation_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with only 1 invalidation signal -- should warn C8."""
d = dict(well_formed_breakout_draft)
d["invalidation_signals"] = ["Breakout fails quickly"]
return d
@pytest.fixture()
def minimal_thesis_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with domain terms but short thesis (~7 words, no mechanism keywords)."""
d = dict(well_formed_breakout_draft)
d["thesis"] = "Breakout with strong volume confirms trend"
return d
@pytest.fixture()
def moderate_thesis_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with domain terms, 12+ words, no mechanism keywords."""
d = dict(well_formed_breakout_draft)
d["thesis"] = (
"Price breakouts above 20-day highs with strong volume and momentum signal trend expansion"
)
return d
@pytest.fixture()
def rich_thesis_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with domain terms, 20+ words, AND mechanism keywords."""
d = dict(well_formed_breakout_draft)
d["thesis"] = (
"Price breakouts above 20-day highs with above-average volume indicate "
"institutional participation and momentum continuation, driven by "
"accumulation and drift dynamics across sessions"
)
return d
@pytest.fixture()
def lean_conditions_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with only 3 conditions + 1 trend filter = 4 total filters."""
d = dict(well_formed_breakout_draft)
d["entry"] = {
"conditions": [
"close > high20_prev",
"rel_volume >= 2",
"close > ma50",
],
"trend_filter": [
"price > sma_200",
],
}
return d
@pytest.fixture()
def moderate_conditions_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with 5 conditions + 2 trend filters = 7 total filters."""
d = dict(well_formed_breakout_draft)
d["entry"] = {
"conditions": [
"close > high20_prev",
"rel_volume >= 2",
"close > ma50",
"close > ma200",
"RSI > 50",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
],
}
return d
@pytest.fixture()
def high_opportunity_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with minimal restrictions -> high est opportunities (~80+)."""
d = dict(well_formed_breakout_draft)
d["regime"] = "Neutral"
d["entry"] = {
"conditions": [
"close > high20_prev",
"rel_volume >= 2",
],
"trend_filter": [
"price > sma_200",
],
}
return d
@pytest.fixture()
def low_opportunity_draft(well_formed_breakout_draft: dict) -> dict:
"""Draft with sector filter + regime + many conditions -> est ~20."""
d = dict(well_formed_breakout_draft)
d["regime"] = "RiskOn"
d["entry"] = {
"conditions": [
"sector == Technology",
"close > high20_prev",
"rel_volume >= 2",
"close > ma50",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
}
return d
@pytest.fixture()
def research_probe_draft() -> dict:
"""A research probe draft that should PASS but not be export eligible."""
return {
"id": "draft_edge_concept_news_reaction_research_probe",
"concept_id": "edge_concept_news_reaction",
"name": "Event overreaction and drift (research_probe)",
"variant": "research_probe",
"export_ready_v1": False,
"entry_family": "research_only",
"hypothesis_type": "news_reaction",
"mechanism_tag": "behavior",
"regime": "Neutral",
"thesis": "Stocks with large negative earnings surprises show post-event drift as institutional investors slowly adjust positions over multiple days.",
"entry": {
"conditions": [
"earnings_surprise < -0.10",
"volume > 2 * avg_volume_50",
],
"trend_filter": [
"price > sma_200",
],
},
"exit": {
"stop_loss_pct": 0.05,
"take_profit_rr": 2.2,
},
"risk": {
"risk_per_trade": 0.005,
"max_positions": 3,
"max_sector_exposure": 0.3,
},
"invalidation_signals": [
"Price recovers within 1 day",
"Volume normalizes immediately",
],
"validation_plan": {
"period": "2016-01-01 to latest",
"entry_timing": "next_open",
"hold_days": [5, 20],
"success_criteria": [
"expected_value_after_costs > 0",
"stable across regimes and subperiods",
],
},
}
scripts/tests/test_review_strategy_drafts.py
"""Unit tests for review_strategy_drafts.py."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
import review_strategy_drafts as rsd
import yaml
CHECKLIST_PATH = Path(__file__).resolve().parents[2] / "assets" / "bias_checklist.yaml"
# ---------------------------------------------------------------------------
# C1: Edge Plausibility
# ---------------------------------------------------------------------------
def test_pass_verdict_well_formed_breakout(well_formed_breakout_draft: dict) -> None:
"""A well-formed breakout draft should PASS review."""
review = rsd.review_draft(well_formed_breakout_draft)
assert review.verdict == "PASS"
assert review.confidence_score >= 70
c1 = _find(review, "C1_edge_plausibility")
assert c1.severity == "pass"
def test_reject_empty_thesis(empty_thesis_draft: dict) -> None:
"""Empty thesis should fail C1 and trigger REJECT."""
review = rsd.review_draft(empty_thesis_draft)
c1 = _find(review, "C1_edge_plausibility")
assert c1.severity == "fail"
assert c1.score == 10
assert review.verdict == "REJECT"
def test_reject_generic_thesis(generic_thesis_draft: dict) -> None:
"""Generic thesis with no domain terms should warn C1."""
c1 = rsd.evaluate_c1(generic_thesis_draft)
assert c1.severity == "warn"
assert c1.score == 40
# ---------------------------------------------------------------------------
# C2: Overfitting Risk
# ---------------------------------------------------------------------------
def test_c2_pass_standard_conditions(well_formed_breakout_draft: dict) -> None:
"""Standard condition count (<= 10) should pass C2."""
review = rsd.review_draft(well_formed_breakout_draft)
c2 = _find(review, "C2_overfitting_risk")
assert c2.severity == "pass"
assert c2.score >= 60
def test_c2_warn_excessive_conditions(excessive_conditions_draft: dict) -> None:
"""11 total conditions should warn C2."""
review = rsd.review_draft(excessive_conditions_draft)
c2 = _find(review, "C2_overfitting_risk")
assert c2.severity == "warn"
assert c2.score == 40
def test_c2_fail_extreme_conditions(extreme_conditions_draft: dict) -> None:
"""13 total conditions should fail C2."""
review = rsd.review_draft(extreme_conditions_draft)
c2 = _find(review, "C2_overfitting_risk")
assert c2.severity == "fail"
assert c2.score == 10
def test_c2_precise_threshold_penalty(precise_threshold_draft: dict) -> None:
"""Precise decimal thresholds should reduce C2 score by 10 each."""
review = rsd.review_draft(precise_threshold_draft)
c2 = _find(review, "C2_overfitting_risk")
# 6 total conditions (pass base 80), but 3 precise thresholds: 80 - 30 = 50
assert c2.score == 50
assert c2.severity == "warn"
# ---------------------------------------------------------------------------
# C3: Sample Adequacy
# ---------------------------------------------------------------------------
def test_c3_pass_broad_conditions(well_formed_breakout_draft: dict) -> None:
"""Broad conditions should pass C3."""
review = rsd.review_draft(well_formed_breakout_draft)
c3 = _find(review, "C3_sample_adequacy")
assert c3.severity == "pass"
def test_c3_warn_restrictive_conditions(restrictive_conditions_draft: dict) -> None:
"""Sector+regime+many conditions should warn C3."""
review = rsd.review_draft(restrictive_conditions_draft)
c3 = _find(review, "C3_sample_adequacy")
assert c3.severity == "warn"
# est ~20 -> _c3_score_from_estimate(20): 10-29 band -> linear 30..59
# (20-10)/(29-10)*(59-30)+30 = 10/19*29+30 ~= 45
assert 40 <= c3.score <= 50
def test_c3_fail_extreme_restriction(extreme_restriction_draft: dict) -> None:
"""Extreme restriction should fail C3."""
review = rsd.review_draft(extreme_restriction_draft)
c3 = _find(review, "C3_sample_adequacy")
assert c3.severity == "fail"
assert c3.score == 10
# ---------------------------------------------------------------------------
# C4: Regime Dependency
# ---------------------------------------------------------------------------
def test_regime_dependency_single_regime(
single_regime_no_validation_draft: dict,
) -> None:
"""Single regime without cross-regime validation should warn C4."""
review = rsd.review_draft(single_regime_no_validation_draft)
c4 = _find(review, "C4_regime_dependency")
assert c4.severity == "warn"
assert c4.score == 40
# ---------------------------------------------------------------------------
# C5: Exit Calibration
# ---------------------------------------------------------------------------
def test_exit_logic_flags_unreasonable_stop(unreasonable_stop_draft: dict) -> None:
"""Stop loss > 15% should fail C5."""
review = rsd.review_draft(unreasonable_stop_draft)
c5 = _find(review, "C5_exit_calibration")
assert c5.severity == "fail"
assert c5.score == 10
def test_exit_logic_flags_low_rr(low_rr_draft: dict) -> None:
"""Risk-reward < 1.5 should fail C5."""
review = rsd.review_draft(low_rr_draft)
c5 = _find(review, "C5_exit_calibration")
assert c5.severity == "fail"
assert c5.score == 10
# ---------------------------------------------------------------------------
# C6: Risk Concentration
# ---------------------------------------------------------------------------
def test_risk_concentration_warn_high_risk(high_risk_draft: dict) -> None:
"""risk_per_trade > 1.5% should warn C6."""
review = rsd.review_draft(high_risk_draft)
c6 = _find(review, "C6_risk_concentration")
assert c6.severity == "warn"
assert c6.score == 40
def test_risk_concentration_fail_extreme(extreme_risk_draft: dict) -> None:
"""risk_per_trade > 2% should fail C6."""
review = rsd.review_draft(extreme_risk_draft)
c6 = _find(review, "C6_risk_concentration")
assert c6.severity == "fail"
assert c6.score == 10
# ---------------------------------------------------------------------------
# C7: Execution Realism
# ---------------------------------------------------------------------------
def test_execution_realism_no_volume_filter(no_volume_filter_draft: dict) -> None:
"""No volume filter in conditions should warn C7."""
review = rsd.review_draft(no_volume_filter_draft)
c7 = _find(review, "C7_execution_realism")
assert c7.severity == "warn"
assert c7.score == 50
def test_c7_fail_export_ready_wrong_family(wrong_family_export_draft: dict) -> None:
"""export_ready=true with non-exportable family should fail C7."""
review = rsd.review_draft(wrong_family_export_draft)
c7 = _find(review, "C7_execution_realism")
assert c7.severity == "fail"
assert c7.score == 10
# ---------------------------------------------------------------------------
# C8: Invalidation Quality
# ---------------------------------------------------------------------------
def test_invalidation_empty_signals(empty_invalidation_draft: dict) -> None:
"""Empty invalidation signals should fail C8."""
review = rsd.review_draft(empty_invalidation_draft)
c8 = _find(review, "C8_invalidation_quality")
assert c8.severity == "fail"
assert c8.score == 10
def test_invalidation_insufficient(insufficient_invalidation_draft: dict) -> None:
"""Only 1 invalidation signal should warn C8."""
review = rsd.review_draft(insufficient_invalidation_draft)
c8 = _find(review, "C8_invalidation_quality")
assert c8.severity == "warn"
assert c8.score == 40
# ---------------------------------------------------------------------------
# Verdict and Confidence
# ---------------------------------------------------------------------------
def test_confidence_score_weighted_average(well_formed_breakout_draft: dict) -> None:
"""Confidence score should be a weighted average of all criteria."""
review = rsd.review_draft(well_formed_breakout_draft)
# C1=95 (continuous: 50+30+10+10), C2=80 (6 filters), C3=~69 (est~39),
# C4=80, C5=80, C6=80, C7=80, C8=80
# Weighted: (20*95+20*80+15*69+10*80+10*80+10*80+10*80+5*80)/100 = 81
expected = 81
assert review.confidence_score == expected
def test_verdict_reject_overrides_on_c1_fail(empty_thesis_draft: dict) -> None:
"""C1 fail should immediately reject regardless of other scores."""
review = rsd.review_draft(empty_thesis_draft)
assert review.verdict == "REJECT"
def test_revision_instructions_populated(generic_thesis_draft: dict) -> None:
"""REVISE verdict should include revision instructions."""
review = rsd.review_draft(generic_thesis_draft)
# generic thesis -> C1 warn -> overall may be REVISE or PASS depending on score
# With C1=40 (warn), weighted: 20*40 + 80*(80) = 800 + 6400 = 7200 / 100 = 72
# Actually let's check: C1=40, all others=80
# (20*40 + 20*80 + 15*80 + 10*80 + 10*80 + 10*80 + 10*80 + 5*80) / 100
# = (800 + 1600 + 1200 + 800 + 800 + 800 + 800 + 400) / 100 = 7200 / 100 = 72
# confidence = 72, no fail -> PASS
# But the C1 finding has revision_instruction since it's warn
c1 = _find(review, "C1_edge_plausibility")
assert c1.revision_instruction is not None
# ---------------------------------------------------------------------------
# Export Eligibility
# ---------------------------------------------------------------------------
def test_export_eligible_pass_and_exportable(
well_formed_breakout_draft: dict,
) -> None:
"""PASS + export_ready_v1 + exportable family -> export_eligible."""
review = rsd.review_draft(well_formed_breakout_draft)
assert review.verdict == "PASS"
assert review.export_eligible is True
def test_export_ineligible_pass_research_only(research_probe_draft: dict) -> None:
"""PASS + research_only family -> not export_eligible."""
review = rsd.review_draft(research_probe_draft)
assert review.export_eligible is False
# ---------------------------------------------------------------------------
# Strict Export Mode
# ---------------------------------------------------------------------------
def test_strict_export_warn_triggers_revise(no_volume_filter_draft: dict) -> None:
"""In strict mode, export-eligible draft with C7 warn → REVISE."""
# no_volume_filter_draft is export_ready_v1=True, entry_family=pivot_breakout
# It gets C7 warn (no volume filter) -> normally still PASS
normal_review = rsd.review_draft(no_volume_filter_draft)
assert normal_review.verdict == "PASS"
strict_review = rsd.review_draft(no_volume_filter_draft, strict_export=True)
assert strict_review.verdict == "REVISE"
assert len(strict_review.revision_instructions) > 0
def test_strict_export_no_warn_still_passes(well_formed_breakout_draft: dict) -> None:
"""In strict mode, export-eligible draft with no warns → still PASS."""
review = rsd.review_draft(well_formed_breakout_draft, strict_export=True)
assert review.verdict == "PASS"
assert review.export_eligible is True
def test_strict_export_research_probe_unaffected(no_volume_filter_draft: dict) -> None:
"""In strict mode, non-export-eligible draft with warns → still PASS (not affected)."""
# Make a research-only version
d = dict(no_volume_filter_draft)
d["export_ready_v1"] = False
d["entry_family"] = "research_only"
review = rsd.review_draft(d, strict_export=True)
# Research probes are not export-eligible, so strict mode does not apply
assert review.verdict == "PASS"
# ---------------------------------------------------------------------------
# CLI and Output
# ---------------------------------------------------------------------------
def test_cli_drafts_dir(tmp_path: Path, well_formed_breakout_draft: dict) -> None:
"""CLI with --drafts-dir should process all YAML files."""
drafts_dir = tmp_path / "drafts"
drafts_dir.mkdir()
(drafts_dir / "draft1.yaml").write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(["--drafts-dir", str(drafts_dir), "--output-dir", str(output_dir)])
assert rc == 0
assert (output_dir / "review.yaml").exists()
def test_cli_single_draft(tmp_path: Path, well_formed_breakout_draft: dict) -> None:
"""CLI with --draft should process a single YAML file."""
draft_file = tmp_path / "single_draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(["--draft", str(draft_file), "--output-dir", str(output_dir)])
assert rc == 0
assert (output_dir / "review.yaml").exists()
def test_cli_format_yaml(tmp_path: Path, well_formed_breakout_draft: dict) -> None:
"""CLI with --format yaml should produce valid YAML output."""
draft_file = tmp_path / "draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--draft",
str(draft_file),
"--output-dir",
str(output_dir),
"--format",
"yaml",
]
)
assert rc == 0
review_path = output_dir / "review.yaml"
assert review_path.exists()
data = yaml.safe_load(review_path.read_text())
assert "summary" in data
assert "reviews" in data
def test_cli_format_json(tmp_path: Path, well_formed_breakout_draft: dict) -> None:
"""CLI with --format json should produce valid JSON output."""
draft_file = tmp_path / "draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--draft",
str(draft_file),
"--output-dir",
str(output_dir),
"--format",
"json",
]
)
assert rc == 0
review_path = output_dir / "review.json"
assert review_path.exists()
data = json.loads(review_path.read_text())
assert "summary" in data
assert "reviews" in data
def test_cli_markdown_summary(tmp_path: Path, well_formed_breakout_draft: dict) -> None:
"""CLI with --markdown-summary should produce a markdown file."""
draft_file = tmp_path / "draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--draft",
str(draft_file),
"--output-dir",
str(output_dir),
"--markdown-summary",
]
)
assert rc == 0
md_path = output_dir / "review_summary.md"
assert md_path.exists()
content = md_path.read_text()
assert "# Strategy Draft Review Summary" in content
# ---------------------------------------------------------------------------
# C1 Continuous Scoring
# ---------------------------------------------------------------------------
def test_c1_pass_score_increases_with_thesis_richness(
minimal_thesis_draft: dict,
moderate_thesis_draft: dict,
rich_thesis_draft: dict,
) -> None:
"""Rich thesis should score higher than moderate, which scores higher than minimal."""
c1_minimal = rsd.evaluate_c1(minimal_thesis_draft)
c1_moderate = rsd.evaluate_c1(moderate_thesis_draft)
c1_rich = rsd.evaluate_c1(rich_thesis_draft)
assert c1_minimal.score < c1_moderate.score < c1_rich.score
def test_c1_mechanism_keyword_bonus(
moderate_thesis_draft: dict,
rich_thesis_draft: dict,
) -> None:
"""Thesis with mechanism keywords should score higher than same-length without."""
c1_no_mechanism = rsd.evaluate_c1(moderate_thesis_draft)
c1_with_mechanism = rsd.evaluate_c1(rich_thesis_draft)
assert c1_with_mechanism.score > c1_no_mechanism.score
def test_c1_pass_score_capped_at_95(rich_thesis_draft: dict) -> None:
"""Even the richest thesis should not exceed 95."""
c1 = rsd.evaluate_c1(rich_thesis_draft)
assert c1.score <= 95
assert c1.severity == "pass"
# ---------------------------------------------------------------------------
# C2 Five-Tier Ordering
# ---------------------------------------------------------------------------
def test_c2_lean_conditions_score_highest(lean_conditions_draft: dict) -> None:
"""4 filters should score 90 (highest tier)."""
c2 = rsd.evaluate_c2(lean_conditions_draft)
assert c2.score == 90
assert c2.severity == "pass"
def test_c2_moderate_conditions_score(moderate_conditions_draft: dict) -> None:
"""7 filters should score 80 (second tier)."""
c2 = rsd.evaluate_c2(moderate_conditions_draft)
assert c2.score == 80
assert c2.severity == "pass"
def test_c2_five_tier_ordering() -> None:
"""Scores should decrease monotonically: 4 < 7 < 10 < 12 < 13 filters."""
def _make_draft(n_conditions: int, n_trend: int) -> dict:
return {
"entry": {
"conditions": [f"cond_{i}" for i in range(n_conditions)],
"trend_filter": [f"trend_{i}" for i in range(n_trend)],
}
}
scores = [
rsd.evaluate_c2(_make_draft(3, 1)).score, # 4 total
rsd.evaluate_c2(_make_draft(5, 2)).score, # 7 total
rsd.evaluate_c2(_make_draft(7, 3)).score, # 10 total
rsd.evaluate_c2(_make_draft(9, 3)).score, # 12 total
rsd.evaluate_c2(_make_draft(10, 3)).score, # 13 total
]
for i in range(len(scores) - 1):
assert scores[i] >= scores[i + 1], (
f"scores[{i}]={scores[i]} < scores[{i + 1}]={scores[i + 1]}"
)
# Verify strict decrease somewhere (not all equal)
assert scores[0] > scores[-1]
# ---------------------------------------------------------------------------
# C3 Continuous Scoring
# ---------------------------------------------------------------------------
def test_c3_high_opportunity_scores_above_75(high_opportunity_draft: dict) -> None:
"""High-opportunity draft should score >= 75."""
c3 = rsd.evaluate_c3(high_opportunity_draft)
assert c3.score >= 75
assert c3.severity == "pass"
def test_c3_low_opportunity_scores_in_warn_band(low_opportunity_draft: dict) -> None:
"""Low-opportunity (~20 est) should score in 30..59 band."""
c3 = rsd.evaluate_c3(low_opportunity_draft)
assert 30 <= c3.score <= 59
assert c3.severity == "warn"
def test_c3_monotonicity() -> None:
"""Higher opportunity estimate should produce higher or equal score."""
estimates = [5, 10, 20, 30, 50, 100, 200, 300]
scores = [rsd._c3_score_from_estimate(e) for e in estimates]
for i in range(len(scores) - 1):
assert scores[i] <= scores[i + 1], (
f"_c3_score_from_estimate({estimates[i]})={scores[i]} > "
f"_c3_score_from_estimate({estimates[i + 1]})={scores[i + 1]}"
)
# ---------------------------------------------------------------------------
# Confidence and Integration
# ---------------------------------------------------------------------------
def test_confidence_score_varies_with_thesis(
minimal_thesis_draft: dict,
rich_thesis_draft: dict,
) -> None:
"""Two drafts with different thesis richness should have different confidence scores."""
review_minimal = rsd.review_draft(minimal_thesis_draft)
review_rich = rsd.review_draft(rich_thesis_draft)
assert review_minimal.confidence_score != review_rich.confidence_score
def test_c1_c2_c3_score_ranges_no_longer_uniform(
rich_thesis_draft: dict,
lean_conditions_draft: dict,
high_opportunity_draft: dict,
) -> None:
"""Verify that modified criteria no longer always produce exactly {10,40,80}."""
old_values = {10, 40, 80}
c1 = rsd.evaluate_c1(rich_thesis_draft)
c2 = rsd.evaluate_c2(lean_conditions_draft)
c3 = rsd.evaluate_c3(high_opportunity_draft)
# At least one of the scores should NOT be in {10, 40, 80}
scores = {c1.score, c2.score, c3.score}
assert not scores.issubset(old_values), f"All scores {scores} still in old set {old_values}"
# ---------------------------------------------------------------------------
# Bias Checklist (issue #297 slice 1)
# ---------------------------------------------------------------------------
@pytest.fixture()
def bias_items() -> list[dict]:
"""Load the bundled bias checklist items."""
return rsd.load_bias_checklist(CHECKLIST_PATH)
def test_load_bias_checklist_parses() -> None:
"""The bundled bias checklist YAML must expose versioned required items."""
items = rsd.load_bias_checklist(CHECKLIST_PATH)
assert len(items) >= 12
assert all({"id", "title", "required", "coverage_hint"} <= set(i) for i in items)
ids = {i["id"] for i in items}
assert {
"look_ahead",
"survivorship",
"universe_selection",
"delisted_securities",
"earnings_announcement_timing",
"split_dividend_adjustment",
"transaction_costs_slippage",
"short_borrow_availability",
"liquidity_capacity",
"parameter_multiplicity",
"walk_forward_out_of_sample",
"benchmark_attribution",
"regime_dependence",
} <= ids
# The issue's checklist is mandatory -> every item is required.
assert all(i["required"] for i in items)
def test_evaluate_bias_checklist_explicit_block(
well_formed_breakout_draft: dict, bias_items: list[dict]
) -> None:
"""A declared truthy entry marks an item addressed; undeclared items are not."""
d = dict(well_formed_breakout_draft)
d["bias_checklist"] = {
"look_ahead": "point-in-time snapshot; no future data used",
"survivorship": "universe includes delisted names",
}
results = rsd.evaluate_bias_checklist(d, bias_items)
by_id = {r["item_id"]: r for r in results}
assert by_id["look_ahead"]["addressed"] is True
assert by_id["survivorship"]["addressed"] is True
assert by_id["universe_selection"]["addressed"] is False
assert by_id["regime_dependence"]["addressed"] is False
def test_bias_noted_in_draft_heuristic(
well_formed_breakout_draft: dict, bias_items: list[dict]
) -> None:
"""The informational 'noted_in_draft' flag matches concept text in the draft."""
d = dict(well_formed_breakout_draft)
results = rsd.evaluate_bias_checklist(d, bias_items)
by_id = {r["item_id"]: r for r in results}
assert by_id["regime_dependence"]["noted_in_draft"] is True
# Not declared in an explicit block, so not 'addressed' even if noted.
assert by_id["regime_dependence"]["addressed"] is False
def test_apply_bias_gate_downgrades_pass_to_revise(bias_items: list[dict]) -> None:
"""Unaddressed required item downgrades PASS -> REVISE and clears export."""
review = rsd.DraftReview(
draft_id="x",
verdict="PASS",
confidence_score=80,
export_eligible=True,
findings=[],
revision_instructions=[],
)
d = {"id": "x", "bias_checklist": {"look_ahead": "point-in-time"}}
results = rsd.evaluate_bias_checklist(d, bias_items)
assert rsd.apply_bias_gate(review, results) is True
assert review.verdict == "REVISE"
assert review.export_eligible is False
def test_apply_bias_gate_no_downgrade_when_all_required_addressed() -> None:
"""When every required item is addressed, the gate leaves the verdict alone."""
items = [
{"id": "a", "title": "A", "required": True, "coverage_hint": "x"},
{"id": "b", "title": "B", "required": True, "coverage_hint": "y"},
]
d = {"id": "x", "bias_checklist": {"a": "done", "b": "done"}}
review = rsd.DraftReview("x", "PASS", 80, True, [], [])
results = rsd.evaluate_bias_checklist(d, items)
assert rsd.apply_bias_gate(review, results) is False
assert review.verdict == "PASS"
assert review.export_eligible is True
def test_apply_bias_gate_never_upgrades_reject(bias_items: list[dict]) -> None:
"""A REJECT verdict must never be raised by the bias gate."""
review = rsd.DraftReview("x", "REJECT", 20, False, [], [])
d = {"id": "x"}
results = rsd.evaluate_bias_checklist(d, bias_items)
assert rsd.apply_bias_gate(review, results) is False
assert review.verdict == "REJECT"
def test_build_output_includes_bias_block_only_when_present(
well_formed_breakout_draft: dict,
) -> None:
"""bias_review is serialized only when the checklist source is supplied."""
review = rsd.review_draft(well_formed_breakout_draft)
bias_block = [
{
"item_id": "look_ahead",
"title": "Look-ahead bias",
"required": True,
"addressed": True,
"noted_in_draft": True,
"note": "point-in-time",
}
]
with_bias = rsd.build_output("src", 1, [review], bias_by_id={review.draft_id: bias_block})
assert with_bias["reviews"][0]["bias_review"] == bias_block
without_bias = rsd.build_output("src", 1, [review])
assert "bias_review" not in without_bias["reviews"][0]
def test_bias_checklist_doc_matches_asset(bias_items: list[dict]) -> None:
"""Drift guard: the research-bias doc must describe every checklist item."""
doc = (
Path(__file__).resolve().parents[4] / "docs/dev/strategy-research-bias-checklist.md"
).read_text(encoding="utf-8")
for item in bias_items:
assert f"`{item['id']}`" in doc, f"bias doc missing checklist item {item['id']}"
def test_cli_bias_checklist_defaults_bundled(
tmp_path: Path, well_formed_breakout_draft: dict
) -> None:
"""--bias-checklist without a path uses the bundled asset and downgrades on gaps."""
draft_file = tmp_path / "draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--draft",
str(draft_file),
"--output-dir",
str(output_dir),
"--format",
"json",
"--bias-checklist",
]
)
assert rc == 0
payload = json.loads((output_dir / "review.json").read_text(encoding="utf-8"))
review = payload["reviews"][0]
assert review["verdict"] == "REVISE"
assert review["export_eligible"] is False
assert "bias_review" in review
assert review["bias_review"]
assert any(not b["addressed"] for b in review["bias_review"])
def test_cli_bias_checklist_custom_path(tmp_path: Path, well_formed_breakout_draft: dict) -> None:
"""--bias-checklist with an explicit path loads that checklist."""
draft_file = tmp_path / "draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
checklist_file = tmp_path / "custom.yaml"
checklist_file.write_text(
"schema_version: '1.0'\nitems:\n - id: look_ahead\n title: Look ahead\n"
" required: true\n coverage_hint: point[- ]in[- ]time\n",
encoding="utf-8",
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--draft",
str(draft_file),
"--output-dir",
str(output_dir),
"--format",
"json",
"--bias-checklist",
str(checklist_file),
]
)
assert rc == 0
review = json.loads((output_dir / "review.json").read_text(encoding="utf-8"))["reviews"][0]
assert review["verdict"] == "REVISE"
assert len(review["bias_review"]) == 1
def test_cli_bias_checklist_missing_file(tmp_path: Path, well_formed_breakout_draft: dict) -> None:
"""A missing custom checklist path surfaces a ReviewError (rc=1)."""
draft_file = tmp_path / "draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--draft",
str(draft_file),
"--output-dir",
str(output_dir),
"--bias-checklist",
str(tmp_path / "nope.yaml"),
]
)
assert rc == 1
def test_cli_bias_checklist_malformed_yaml(
tmp_path: Path, well_formed_breakout_draft: dict
) -> None:
"""A custom checklist with invalid YAML surfaces a ReviewError (rc=1), not a traceback."""
draft_file = tmp_path / "draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
checklist_file = tmp_path / "bad.yaml"
checklist_file.write_text("items: [unclosed\n", encoding="utf-8")
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--draft",
str(draft_file),
"--output-dir",
str(output_dir),
"--bias-checklist",
str(checklist_file),
]
)
assert rc == 1
def test_cli_bias_checklist_schema_version_mismatch(
tmp_path: Path, well_formed_breakout_draft: dict
) -> None:
"""A custom checklist with a mismatched schema_version is rejected."""
draft_file = tmp_path / "draft.yaml"
draft_file.write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
checklist_file = tmp_path / "checklist.yaml"
checklist_file.write_text(
yaml.safe_dump(
{"schema_version": "2.0", "items": [{"id": "x", "coverage_hint": "x"}]},
sort_keys=False,
),
encoding="utf-8",
)
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--draft",
str(draft_file),
"--output-dir",
str(output_dir),
"--bias-checklist",
str(checklist_file),
]
)
assert rc == 1
def test_cli_bias_checklist_drafts_dir_joins_by_index(
tmp_path: Path, well_formed_breakout_draft: dict
) -> None:
"""--drafts-dir mode: a PASS draft is downgraded, a REJECT draft untouched."""
drafts_dir = tmp_path / "drafts"
drafts_dir.mkdir()
(drafts_dir / "a.yaml").write_text(
yaml.safe_dump(well_formed_breakout_draft, sort_keys=False), encoding="utf-8"
)
rejecting = dict(well_formed_breakout_draft)
rejecting["id"] = "draft_rejected_bad_idea"
rejecting["thesis"] = ""
(drafts_dir / "b.yaml").write_text(yaml.safe_dump(rejecting, sort_keys=False), encoding="utf-8")
output_dir = tmp_path / "output"
output_dir.mkdir()
rc = rsd.main(
[
"--drafts-dir",
str(drafts_dir),
"--output-dir",
str(output_dir),
"--format",
"json",
"--bias-checklist",
]
)
assert rc == 0
reviews = json.loads((output_dir / "review.json").read_text(encoding="utf-8"))["reviews"]
assert len(reviews) == 2
pass_review = next(r for r in reviews if r["draft_id"] == well_formed_breakout_draft["id"])
reject_review = next(r for r in reviews if r["draft_id"] == "draft_rejected_bad_idea")
assert pass_review["verdict"] == "REVISE"
assert "bias_review" in pass_review and pass_review["bias_review"]
assert any("Declare how" in i for i in pass_review["revision_instructions"])
assert reject_review["verdict"] == "REJECT"
assert "bias_review" in reject_review
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _find(review: rsd.DraftReview, criterion: str) -> rsd.ReviewFinding:
"""Find a specific criterion finding in a review."""
for f in review.findings:
if f.criterion == criterion:
return f
raise AssertionError(f"Criterion {criterion} not found in review findings")
SKILL.md
---
name: edge-strategy-reviewer
description: >
Critically review strategy drafts from edge-strategy-designer for edge
plausibility, overfitting risk, sample size adequacy, and execution realism.
Use when strategy_drafts/*.yaml exists and needs quality gate before pipeline
export. Outputs PASS/REVISE/REJECT verdicts with confidence scores.
---
# Edge Strategy Reviewer
Deterministic quality gate for strategy drafts produced by `edge-strategy-designer`.
## When to Use
- After `edge-strategy-designer` generates `strategy_drafts/*.yaml`
- Before exporting drafts to `edge-candidate-agent` via the pipeline
- When manually validating a draft strategy for edge plausibility
## Prerequisites
- Strategy draft YAML files (output of `edge-strategy-designer`)
- Python 3.10+ with PyYAML
## Workflow
1. Load draft YAML files from `--drafts-dir` or a single `--draft` file
2. Evaluate each draft against 8 criteria (C1-C8) with weighted scoring
3. Compute confidence score (weighted average of all criteria)
4. Determine verdict: PASS / REVISE / REJECT
5. Assess export eligibility (PASS + export_ready_v1 + exportable family)
6. Write review output (YAML or JSON) and optional markdown summary
## Review Criteria
| # | Criterion | Weight | Key Checks |
|---|-----------|--------|------------|
| C1 | Edge Plausibility | 20 | Thesis quality, domain terms, mechanism keywords (continuous 50-95) |
| C2 | Overfitting Risk | 20 | 5-tier filter count scoring (90/80/60/40/10), precise threshold penalty |
| C3 | Sample Adequacy | 15 | Continuous scoring from estimated annual opportunities (10-95) |
| C4 | Regime Dependency | 10 | Cross-regime validation |
| C5 | Exit Calibration | 10 | Stop-loss, reward-to-risk |
| C6 | Risk Concentration | 10 | Position sizing limits |
| C7 | Execution Realism | 10 | Volume filter, export consistency |
| C8 | Invalidation Quality | 5 | Signal count and specificity |
## Verdict Logic
- C1 or C2 severity=fail → immediate REJECT
- confidence >= 70, no fail findings → PASS
- confidence < 35 → REJECT
- Otherwise → REVISE (with revision instructions)
## Running the Script
```bash
# Review all drafts in a directory
python3 skills/edge-strategy-reviewer/scripts/review_strategy_drafts.py \
--drafts-dir reports/edge_strategy_drafts/ \
--output-dir reports/
# Single draft review
python3 skills/edge-strategy-reviewer/scripts/review_strategy_drafts.py \
--draft reports/edge_strategy_drafts/draft_xxx.yaml \
--output-dir reports/
# JSON output with markdown summary
python3 skills/edge-strategy-reviewer/scripts/review_strategy_drafts.py \
--drafts-dir reports/edge_strategy_drafts/ \
--output-dir reports/ \
--format json \
--markdown-summary
# Strict export mode: export-eligible drafts with any warn → REVISE
python3 skills/edge-strategy-reviewer/scripts/review_strategy_drafts.py \
--drafts-dir reports/edge_strategy_drafts/ \
--output-dir reports/ \
--strict-export
# Bias checklist (issue #297): mandatory strategy-research bias gate.
# Uses the bundled checklist unless a custom YAML path is supplied.
# A draft leaving any required item unaddressed is downgraded PASS → REVISE
# and loses export eligibility.
python3 skills/edge-strategy-reviewer/scripts/review_strategy_drafts.py \
--drafts-dir reports/edge_strategy_drafts/ \
--output-dir reports/ \
--bias-checklist
```
## Output Format
Primary output: `review.yaml` (or `review.json`)
```yaml
generated_at_utc: "2026-02-28T12:00:00+00:00"
source:
drafts_dir: "/path/to/strategy_drafts"
draft_count: 4
summary:
total: 4
PASS: 1
REVISE: 2
REJECT: 1
export_eligible: 1
reviews:
- draft_id: "draft_xxx_core"
verdict: "PASS"
confidence_score: 80
export_eligible: true
findings: [...]
revision_instructions: []
```
## Resources
- `references/review_criteria.md` — Detailed scoring rubric for C1-C8
- `references/overfitting_checklist.md` — Overfitting detection heuristics