scripts/design_strategy_drafts.py
#!/usr/bin/env python3
"""Design strategy drafts from abstract edge concepts."""
from __future__ import annotations
import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import yaml
DEFAULT_EXPORTABLE_FAMILIES = {
"pivot_breakout",
"gap_up_continuation",
"panic_reversal",
"news_reaction",
}
ENTRY_TEMPLATE = {
"pivot_breakout": {
"conditions": [
"close > high20_prev",
"rel_volume >= 1.5",
"close > ma50 > ma200",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
},
"gap_up_continuation": {
"conditions": [
"gap_up_detected",
"close_above_gap_day_high",
"volume > 2.0 * avg_volume_50",
],
"trend_filter": [
"price > sma_200",
"price > sma_50",
"sma_50 > sma_200",
],
},
"panic_reversal": {
"conditions": [
"ret_1d <= -0.07",
"rel_volume >= 1.8",
"close > 0.85 * ma200",
],
"trend_filter": [
"price > sma_200 * 0.85",
"no_fresh_breakdown_below_ma200",
],
},
"news_reaction": {
"conditions": [
"abs_reaction_1d >= 0.06",
"rel_volume >= 2.0",
"close_pos >= 0.4",
],
"trend_filter": [
"validate_follow_through_d2",
"volume_confirmation_present",
],
},
}
RISK_PROFILES = {
"conservative": {
"risk_per_trade": 0.005,
"max_positions": 3,
"stop_loss_pct": 0.05,
"take_profit_rr": 2.2,
},
"balanced": {
"risk_per_trade": 0.01,
"max_positions": 5,
"stop_loss_pct": 0.07,
"take_profit_rr": 3.0,
},
"aggressive": {
"risk_per_trade": 0.015,
"max_positions": 7,
"stop_loss_pct": 0.09,
"take_profit_rr": 3.5,
},
}
HYPOTHESIS_EXIT_OVERRIDES: dict[str, dict[str, Any]] = {
"breakout": {
"stop_loss_multiplier": 0.85,
"rr_adjustment": 0.0,
"time_stop_days": 20,
"trailing_stop_hint": "ATR trailing recommended",
},
"futures_trigger": {
"stop_loss_multiplier": 0.85,
"rr_adjustment": 0.0,
"time_stop_days": 20,
"trailing_stop_hint": "ATR trailing recommended",
},
"earnings_drift": {
"stop_loss_multiplier": 1.15,
"rr_adjustment": -0.5,
"time_stop_days": 10,
"trailing_stop_hint": None,
},
"panic_reversal": {
"stop_loss_multiplier": 0.70,
"rr_adjustment": -0.7,
"time_stop_days": 5,
"trailing_stop_hint": None,
},
"news_reaction": {
"stop_loss_multiplier": 0.85,
"rr_adjustment": -0.5,
"time_stop_days": 7,
"trailing_stop_hint": None,
},
"regime_shift": {
"stop_loss_multiplier": 1.0,
"rr_adjustment": 0.0,
"time_stop_days": 15,
"trailing_stop_hint": None,
},
"sector_x_stock": {
"stop_loss_multiplier": 1.0,
"rr_adjustment": 0.0,
"time_stop_days": 15,
"trailing_stop_hint": None,
},
"calendar_anomaly": {
"stop_loss_multiplier": 0.85,
"rr_adjustment": -0.5,
"time_stop_days": 7,
"trailing_stop_hint": None,
},
}
RR_FLOOR = 1.5
VARIANT_OVERRIDES = {
"core": {
"entry_filter_note": "Use baseline confirmation and trend filter.",
"risk_multiplier": 1.0,
},
"conservative": {
"entry_filter_note": "Require stricter confirmation before entry.",
"risk_multiplier": 0.75,
},
"research_probe": {
"entry_filter_note": "Probe setup with small size for hypothesis validation.",
"risk_multiplier": 0.5,
},
}
class StrategyDesignError(Exception):
"""Raised when strategy design cannot proceed."""
def sanitize_identifier(value: str) -> str:
"""Convert free text into a safe identifier."""
lowered = "".join(ch.lower() if ch.isalnum() else "_" for ch in value)
compact = "_".join(part for part in lowered.split("_") if part)
return compact or "draft"
def safe_float(value: Any, default: float = 0.0) -> float:
"""Best-effort float conversion."""
try:
return float(value)
except (TypeError, ValueError):
return default
def apply_hypothesis_exit_overrides(
base_stop: float,
base_rr: float,
hypothesis_type: str,
) -> dict[str, Any]:
"""Apply hypothesis-type-specific exit calibration."""
fallback: dict[str, Any] = {
"stop_loss_multiplier": 1.0,
"rr_adjustment": 0.0,
"time_stop_days": 10,
"trailing_stop_hint": None,
}
overrides = HYPOTHESIS_EXIT_OVERRIDES.get(hypothesis_type, fallback)
stop_loss_pct = round(base_stop * overrides["stop_loss_multiplier"], 4)
take_profit_rr = round(max(RR_FLOOR, base_rr + overrides["rr_adjustment"]), 2)
result: dict[str, Any] = {
"stop_loss_pct": stop_loss_pct,
"take_profit_rr": take_profit_rr,
"time_stop_days": overrides["time_stop_days"],
}
if overrides.get("trailing_stop_hint"):
result["trailing_stop_hint"] = overrides["trailing_stop_hint"]
return result
def load_concepts(path: Path) -> list[dict[str, Any]]:
"""Load concept YAML payload."""
payload = yaml.safe_load(path.read_text())
if not isinstance(payload, dict):
raise StrategyDesignError("concept file must be a mapping")
concepts = payload.get("concepts", [])
if not isinstance(concepts, list):
raise StrategyDesignError("concept file must contain concepts list")
out: list[dict[str, Any]] = []
for concept in concepts:
if isinstance(concept, dict) and concept.get("id"):
out.append(concept)
return out
def resolve_variants(
concept: dict[str, Any], variants_per_concept: int, exportable_families: set[str] | None = None
) -> list[str]:
"""Choose draft variants for a concept."""
strategy_design = concept.get("strategy_design", {})
export_ready = bool(strategy_design.get("export_ready_v1"))
# Re-check recommended family against override when provided
if export_ready and exportable_families is not None:
recommended = strategy_design.get("recommended_entry_family")
if recommended not in exportable_families:
export_ready = False
if export_ready:
variants = ["core", "conservative"]
else:
variants = ["research_probe"]
return variants[: max(variants_per_concept, 1)]
def resolve_entry_settings(concept: dict[str, Any]) -> tuple[str, dict[str, Any], bool]:
"""Resolve entry family and default conditions."""
strategy_design = concept.get("strategy_design", {})
family = strategy_design.get("recommended_entry_family")
if isinstance(family, str) and family in ENTRY_TEMPLATE:
return family, ENTRY_TEMPLATE[family], True
return "research_only", {"conditions": [], "trend_filter": []}, False
def build_draft(
concept: dict[str, Any],
variant: str,
risk_profile: str,
as_of: str | None,
exportable_families: set[str] | None = None,
) -> dict[str, Any]:
"""Build one strategy draft from concept + variant."""
concept_id = str(concept.get("id"))
hypothesis_type = str(concept.get("hypothesis_type", "unknown"))
mechanism_tag = str(concept.get("mechanism_tag", "uncertain"))
regime = str(concept.get("regime", "Unknown"))
support = concept.get("support", {}) if isinstance(concept.get("support"), dict) else {}
abstraction = (
concept.get("abstraction", {}) if isinstance(concept.get("abstraction"), dict) else {}
)
entry_family, entry_template, export_ready = resolve_entry_settings(concept)
representative_conditions = support.get("representative_conditions", [])
if isinstance(representative_conditions, list):
merged_conditions = [
c for c in representative_conditions if isinstance(c, str) and c.strip()
]
else:
merged_conditions = []
if not merged_conditions:
merged_conditions = list(entry_template.get("conditions", []))
trend_filter = entry_template.get("trend_filter", [])
if not isinstance(trend_filter, list):
trend_filter = []
risk_base = RISK_PROFILES[risk_profile]
variant_override = VARIANT_OVERRIDES[variant]
multiplier = safe_float(variant_override.get("risk_multiplier"), 1.0)
risk_per_trade = round(risk_base["risk_per_trade"] * multiplier, 4)
max_positions = max(int(round(risk_base["max_positions"] * multiplier)), 1)
exit_overrides = apply_hypothesis_exit_overrides(
base_stop=risk_base["stop_loss_pct"],
base_rr=risk_base["take_profit_rr"],
hypothesis_type=hypothesis_type,
)
draft_id = sanitize_identifier(f"draft_{concept_id}_{variant}")
return {
"id": draft_id,
"as_of": as_of,
"concept_id": concept_id,
"variant": variant,
"risk_profile": risk_profile,
"name": f"{concept.get('title', concept_id)} ({variant})",
"hypothesis_type": hypothesis_type,
"mechanism_tag": mechanism_tag,
"regime": regime,
"export_ready_v1": bool(
export_ready
and entry_family
in (
exportable_families
if exportable_families is not None
else DEFAULT_EXPORTABLE_FAMILIES
)
),
"entry_family": entry_family,
"entry": {
"conditions": merged_conditions,
"trend_filter": trend_filter,
"note": variant_override["entry_filter_note"],
},
"exit": exit_overrides,
"risk": {
"position_sizing": "fixed_risk",
"risk_per_trade": risk_per_trade,
"max_positions": max_positions,
"max_sector_exposure": 0.3,
},
"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",
"passes pipeline phase1 gates",
],
},
"thesis": str(abstraction.get("thesis", "")),
"invalidation_signals": abstraction.get("invalidation_signals", []),
}
def build_export_ticket(draft: dict[str, Any]) -> dict[str, Any]:
"""Build exportable ticket from strategy draft."""
ticket_id = sanitize_identifier(draft["id"].replace("draft_", "edge_"))
entry_family = draft["entry_family"]
hypothesis_type = str(draft.get("hypothesis_type", "unknown"))
if entry_family == "pivot_breakout" and hypothesis_type == "unknown":
hypothesis_type = "breakout"
if entry_family == "gap_up_continuation" and hypothesis_type == "unknown":
hypothesis_type = "earnings_drift"
return {
"id": ticket_id,
"name": draft["name"],
"description": f"Draft-derived ticket from concept {draft['concept_id']} ({draft['variant']}).",
"hypothesis_type": hypothesis_type,
"entry_family": entry_family,
"mechanism_tag": draft.get("mechanism_tag", "uncertain"),
"regime": draft.get("regime", "Neutral"),
"holding_horizon": "20D",
"entry": {
"conditions": draft.get("entry", {}).get("conditions", []),
"trend_filter": draft.get("entry", {}).get("trend_filter", []),
},
"risk": draft.get("risk", {}),
"exit": {
"stop_loss_pct": draft.get("exit", {}).get("stop_loss_pct", 0.07),
"take_profit_rr": draft.get("exit", {}).get("take_profit_rr", 3.0),
},
"cost_model": {
"commission_per_share": 0.0,
"slippage_bps": 5,
},
}
def parse_args() -> argparse.Namespace:
"""Parse CLI args."""
parser = argparse.ArgumentParser(
description="Design strategy drafts from edge concepts.",
)
parser.add_argument("--concepts", required=True, help="Path to edge_concepts.yaml")
parser.add_argument(
"--output-dir",
default="reports/edge_strategy_drafts",
help="Output directory for strategy draft YAML files",
)
parser.add_argument(
"--risk-profile",
default="balanced",
choices=sorted(RISK_PROFILES.keys()),
help="Risk profile applied to generated drafts",
)
parser.add_argument(
"--variants-per-concept",
type=int,
default=2,
help="Maximum variants to generate per concept",
)
parser.add_argument(
"--exportable-tickets-dir",
default=None,
help="Optional directory to write exportable ticket YAML files",
)
parser.add_argument(
"--exportable-families",
default=None,
help="Comma-separated list of exportable entry families (overrides module default)",
)
return parser.parse_args()
def main() -> int:
"""CLI entrypoint."""
args = parse_args()
concepts_path = Path(args.concepts).resolve()
output_dir = Path(args.output_dir).resolve()
exportable_tickets_dir = (
Path(args.exportable_tickets_dir).resolve() if args.exportable_tickets_dir else None
)
ef_override: set[str] | None = None
if args.exportable_families:
ef_override = {f.strip() for f in args.exportable_families.split(",") if f.strip()}
if not concepts_path.exists():
print(f"[ERROR] concepts file not found: {concepts_path}")
return 1
try:
concepts = load_concepts(concepts_path)
if not concepts:
raise StrategyDesignError("no concepts available in file")
source_payload = yaml.safe_load(concepts_path.read_text())
as_of = source_payload.get("as_of") if isinstance(source_payload, dict) else None
output_dir.mkdir(parents=True, exist_ok=True)
if exportable_tickets_dir is not None:
exportable_tickets_dir.mkdir(parents=True, exist_ok=True)
draft_count = 0
exportable_ticket_count = 0
for concept in concepts:
variants = resolve_variants(
concept=concept,
variants_per_concept=args.variants_per_concept,
exportable_families=ef_override,
)
for variant in variants:
draft = build_draft(
concept=concept,
variant=variant,
risk_profile=args.risk_profile,
as_of=as_of,
exportable_families=ef_override,
)
draft_path = output_dir / f"{draft['id']}.yaml"
draft_path.write_text(yaml.safe_dump(draft, sort_keys=False), encoding="utf-8")
draft_count += 1
if draft.get("export_ready_v1") and exportable_tickets_dir is not None:
ticket = build_export_ticket(draft)
ticket_path = exportable_tickets_dir / f"{ticket['id']}.yaml"
ticket_path.write_text(
yaml.safe_dump(ticket, sort_keys=False), encoding="utf-8"
)
exportable_ticket_count += 1
manifest = {
"generated_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
"concepts_file": str(concepts_path),
"risk_profile": args.risk_profile,
"draft_count": draft_count,
"exportable_ticket_count": exportable_ticket_count,
"output_dir": str(output_dir),
"exportable_tickets_dir": str(exportable_tickets_dir)
if exportable_tickets_dir
else None,
}
(output_dir / "run_manifest.json").write_text(
json.dumps(manifest, indent=2) + "\n", encoding="utf-8"
)
except StrategyDesignError as exc:
print(f"[ERROR] {exc}")
return 1
print(
"[OK] "
f"drafts={draft_count} exportable_tickets={exportable_ticket_count} "
f"output_dir={output_dir}"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/tests/test_design_strategy_drafts.py
"""Unit tests for design_strategy_drafts.py."""
from __future__ import annotations
import design_strategy_drafts as dsd
def _make_concept(hypothesis_type: str, entry_family: str | None, export: bool) -> dict:
"""Create a minimal concept for exit calibration tests."""
return {
"id": f"edge_concept_{hypothesis_type}_test",
"title": f"Test {hypothesis_type}",
"hypothesis_type": hypothesis_type,
"mechanism_tag": "behavior",
"regime": "RiskOn",
"support": {"representative_conditions": ["close > high20_prev", "rel_volume >= 1.5"]},
"abstraction": {"thesis": "Test thesis", "invalidation_signals": ["Signal 1"]},
"strategy_design": {
"recommended_entry_family": entry_family,
"export_ready_v1": export,
},
}
# --- Exit calibration tests ---
def test_exit_calibration_breakout_balanced() -> None:
"""Breakout + balanced: stop=0.07*0.85=0.0595, RR=3.0+0.0=3.0, time=20."""
concept = _make_concept("breakout", "pivot_breakout", export=True)
draft = dsd.build_draft(
concept=concept, variant="core", risk_profile="balanced", as_of="2026-02-28"
)
assert abs(draft["exit"]["stop_loss_pct"] - 0.0595) < 0.001
assert draft["exit"]["take_profit_rr"] == 3.0
assert draft["exit"]["time_stop_days"] == 20
assert "trailing_stop_hint" in draft["exit"]
def test_exit_calibration_earnings_drift_balanced() -> None:
"""Earnings drift + balanced: stop=0.07*1.15=0.0805, RR=3.0-0.5=2.5, time=10."""
concept = _make_concept("earnings_drift", "gap_up_continuation", export=True)
draft = dsd.build_draft(
concept=concept, variant="core", risk_profile="balanced", as_of="2026-02-28"
)
assert abs(draft["exit"]["stop_loss_pct"] - 0.0805) < 0.001
assert draft["exit"]["take_profit_rr"] == 2.5
assert draft["exit"]["time_stop_days"] == 10
assert "trailing_stop_hint" not in draft["exit"]
def test_exit_calibration_panic_reversal_balanced() -> None:
"""Panic reversal + balanced: stop=0.07*0.70=0.049, RR=3.0-0.7=2.3, time=5."""
concept = _make_concept("panic_reversal", None, export=False)
draft = dsd.build_draft(
concept=concept, variant="research_probe", risk_profile="balanced", as_of="2026-02-28"
)
assert abs(draft["exit"]["stop_loss_pct"] - 0.049) < 0.001
assert draft["exit"]["take_profit_rr"] == 2.3
assert draft["exit"]["time_stop_days"] == 5
def test_exit_rr_floor_clamp_conservative_panic() -> None:
"""Conservative(RR=2.2) + panic_reversal(adj=-0.7): 2.2-0.7=1.5 >= 1.5, OK."""
concept = _make_concept("panic_reversal", None, export=False)
draft = dsd.build_draft(
concept=concept, variant="research_probe", risk_profile="conservative", as_of="2026-02-28"
)
assert draft["exit"]["take_profit_rr"] >= dsd.RR_FLOOR # 1.5
def test_exit_calibration_aggressive_breakout() -> None:
"""Aggressive + breakout: stop=0.09*0.85=0.0765, RR=3.5+0.0=3.5."""
concept = _make_concept("breakout", "pivot_breakout", export=True)
draft = dsd.build_draft(
concept=concept, variant="core", risk_profile="aggressive", as_of="2026-02-28"
)
assert abs(draft["exit"]["stop_loss_pct"] - 0.0765) < 0.001
assert draft["exit"]["take_profit_rr"] == 3.5
def test_exit_calibration_unknown_hypothesis_fallback() -> None:
"""Unknown hypothesis_type uses fallback (multiplier=1.0, adj=0.0, time=10)."""
concept = _make_concept("unknown_type", None, export=False)
draft = dsd.build_draft(
concept=concept, variant="research_probe", risk_profile="balanced", as_of="2026-02-28"
)
assert draft["exit"]["stop_loss_pct"] == 0.07 # 0.07 * 1.0
assert draft["exit"]["take_profit_rr"] == 3.0 # 3.0 + 0.0
assert draft["exit"]["time_stop_days"] == 10
def test_export_ticket_preserves_calibrated_exit() -> None:
"""Export ticket should carry the calibrated exit parameters."""
concept = _make_concept("breakout", "pivot_breakout", export=True)
draft = dsd.build_draft(
concept=concept, variant="core", risk_profile="balanced", as_of="2026-02-28"
)
ticket = dsd.build_export_ticket(draft)
assert abs(ticket["exit"]["stop_loss_pct"] - 0.0595) < 0.001
assert ticket["exit"]["take_profit_rr"] == 3.0
def test_c5_bounds_all_combinations() -> None:
"""All hypothesis x profile combinations should have RR >= 1.5 and stop <= 0.15."""
for hyp_type in list(dsd.HYPOTHESIS_EXIT_OVERRIDES.keys()) + ["unknown_xyz"]:
for profile_name, profile in dsd.RISK_PROFILES.items():
result = dsd.apply_hypothesis_exit_overrides(
base_stop=profile["stop_loss_pct"],
base_rr=profile["take_profit_rr"],
hypothesis_type=hyp_type,
)
assert result["take_profit_rr"] >= dsd.RR_FLOOR, f"{hyp_type}/{profile_name} RR too low"
assert result["stop_loss_pct"] <= 0.15, f"{hyp_type}/{profile_name} stop too high"
def test_trailing_stop_hint_present_for_breakout() -> None:
"""Breakout should include trailing_stop_hint, panic_reversal should not."""
result_breakout = dsd.apply_hypothesis_exit_overrides(0.07, 3.0, "breakout")
assert "trailing_stop_hint" in result_breakout
result_panic = dsd.apply_hypothesis_exit_overrides(0.07, 3.0, "panic_reversal")
assert "trailing_stop_hint" not in result_panic
def test_existing_tests_still_pass() -> None:
"""Re-run existing test scenarios to confirm no regression."""
concept = {
"id": "edge_concept_breakout_behavior_riskon",
"title": "Participation-backed trend breakout",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"support": {"representative_conditions": ["close > high20_prev", "rel_volume >= 1.5"]},
"abstraction": {
"thesis": "Breakout thesis",
"invalidation_signals": ["Breakout fails quickly"],
},
"strategy_design": {"recommended_entry_family": "pivot_breakout", "export_ready_v1": True},
}
draft = dsd.build_draft(
concept=concept, variant="core", risk_profile="balanced", as_of="2026-02-20"
)
assert draft["export_ready_v1"] is True
assert draft["entry_family"] == "pivot_breakout"
# --- Original tests ---
def test_build_draft_export_ready_breakout() -> None:
concept = {
"id": "edge_concept_breakout_behavior_riskon",
"title": "Participation-backed trend breakout",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"support": {
"representative_conditions": ["close > high20_prev", "rel_volume >= 1.5"],
},
"abstraction": {
"thesis": "Breakout thesis",
"invalidation_signals": ["Breakout fails quickly"],
},
"strategy_design": {
"recommended_entry_family": "pivot_breakout",
"export_ready_v1": True,
},
}
draft = dsd.build_draft(
concept=concept,
variant="core",
risk_profile="balanced",
as_of="2026-02-20",
)
assert draft["export_ready_v1"] is True
assert draft["entry_family"] == "pivot_breakout"
assert draft["risk_profile"] == "balanced"
assert draft["risk"]["risk_per_trade"] == 0.01
ticket = dsd.build_export_ticket(draft)
assert ticket["entry_family"] == "pivot_breakout"
assert ticket["id"].startswith("edge_")
def test_build_draft_research_probe_for_non_exportable_concept() -> None:
concept = {
"id": "edge_concept_news_reaction_behavior_riskon",
"title": "Event overreaction and drift",
"hypothesis_type": "news_reaction",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"support": {"representative_conditions": ["reaction_1d=-0.132"]},
"abstraction": {},
"strategy_design": {
"recommended_entry_family": None,
"export_ready_v1": False,
},
}
draft = dsd.build_draft(
concept=concept,
variant="research_probe",
risk_profile="conservative",
as_of="2026-02-20",
)
assert draft["entry_family"] == "research_only"
assert draft["export_ready_v1"] is False
assert draft["risk_profile"] == "conservative"