Trade Hypothesis Ideator
- 820 installs
- 2.6k repo stars
- Updated August 4, 2026
- tradermonty/claude-trading-skills
trade-hypothesis-ideator is a trading agent skill that turns trade logs, journal notes, and market data into ranked, falsifiable strategy hypotheses with experiment designs, kill criteria, and optional strategy.yaml expo
About
trade-hypothesis-ideator from tradermonty/claude-trading-skills generates structured trading research from normalized input bundles containing market data, trade logs, and journal snippets. The workflow produces one to five hypothesis cards, critiques and ranks them, and can export pursue-worthy cards into strategy.yaml plus metadata.json artifacts compatible with edge-finder-candidate/v1. Each card emphasizes falsifiable claims, experiment design, and kill criteria so developers discard weak edges early. Reach for trade-hypothesis-ideator after collecting observations that hint at a market edge but before coding a full backtest stack. It suits systematic traders and quant-minded engineers documenting discretionary insights who want YAML-ready strategy candidates. Skip it when you need live execution, broker connectivity, or purely qualitative macro narratives without structured inputs or export schemas.
- Converts unstructured trade logs and journal snippets into 1-5 structured hypothesis cards
- Applies a two-pass workflow: normalization then critique-and-rank
- Outputs experiment designs with explicit kill criteria for each hypothesis
- Optional export to strategy.yaml and metadata.json compatible with edge-finder-candidate
- Pure local calculation — no external API keys required
Trade Hypothesis Ideator by the numbers
- 820 all-time installs (skills.sh)
- +93 installs in the week ending Aug 5, 2026 (Skillselion tracking)
- Ranked #571 of 3,282 Productivity & Planning skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/tradermonty/claude-trading-skills --skill trade-hypothesis-ideatorAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 820 |
|---|---|
| repo stars | ★ 2.6k |
| Last updated | August 4, 2026 |
| Repository | tradermonty/claude-trading-skills ↗ |
How do you turn trade journals into testable hypotheses?
Turn scattered trade logs, journal notes, and market data into ranked, falsifiable strategy hypotheses with built-in experiment designs and kill criteria.
Who is it for?
Systematic traders with structured trade logs and journal snippets who need ranked, falsifiable hypotheses before backtesting code.
Skip if: Live order execution, broker API wiring, or unstructured market commentary without normalized input bundles.
When should I use this skill?
User has trade logs, journal entries, or market observations and wants hypothesis cards, kill criteria, or strategy.yaml export.
What you get
Ranked hypothesis cards, experiment designs, kill criteria, and optional edge-finder-candidate/v1 strategy.yaml plus metadata.json exports.
- ranked hypothesis cards
- strategy.yaml exports
- metadata.json experiment metadata
By the numbers
- Generates 1-5 structured hypothesis cards per run
- Exports pursue cards to strategy.yaml and metadata.json
- Uses edge-finder-candidate/v1 compatible export schema
Files
Trade Hypothesis Ideator
Generate 1-5 structured hypothesis cards from a normalized input bundle, critique and rank them, then optionally export pursue cards into strategy.yaml + metadata.json artifacts.
When to Use
- After gathering trade logs, journal entries, or market observations that suggest a potential edge
- When you have a structured input bundle (JSON) with evidence snippets and want falsifiable hypotheses
- To bridge qualitative observations into quantitative experiment designs
- Before committing capital to validate a new strategy idea with kill criteria
Prerequisites
- Input JSON bundle with one or more of:
trade_log,journal_snippets,market_data,observations - Python 3.9+ with
pyyamlinstalled - No external API keys required (pure calculation skill)
Workflow
1. Receive input JSON bundle. 2. Run pass 1 normalization + evidence extraction. 3. Generate hypotheses with prompts:
prompts/system_prompt.mdprompts/developer_prompt_template.md(inject{{evidence_summary}})
4. Critique hypotheses with prompts/critique_prompt_template.md. 5. Run pass 2 ranking + output formatting + guardrails. 6. Optionally export pursue hypotheses via Step H strategy exporter.
Scripts
- Pass 1 (evidence summary):
python3 skills/trade-hypothesis-ideator/scripts/run_hypothesis_ideator.py \
--input skills/trade-hypothesis-ideator/examples/example_input.json \
--output-dir reports/- Pass 2 (rank + output + optional export):
python3 skills/trade-hypothesis-ideator/scripts/run_hypothesis_ideator.py \
--input skills/trade-hypothesis-ideator/examples/example_input.json \
--hypotheses reports/raw_hypotheses.json \
--output-dir reports/ \
--export-strategiesOutput
hypothesis_cards_<date>.json— Ranked hypothesis cards with verdicts (pursue,revise,discard)hypothesis_cards_<date>.md— Human-readable summary with experiment designs and kill criteriastrategy_<hypothesis_id>.yaml— (Optional) Edge-finder-compatible strategy export forpursuecardsmetadata_<hypothesis_id>.json— (Optional) Provenance metadata for exported strategies
Resources
references/hypothesis_types.md— Taxonomy of hypothesis patterns (mean-reversion, momentum, event-driven, etc.)references/evidence_quality_guide.md— Criteria for rating evidence strength and sample size requirements
{
"objective": {
"goal": "Improve consistency of an earnings gap-up continuation setup",
"focus_area": "Entry timing and exit calibration",
"horizon": "4-8 weeks"
},
"strategy_context": {
"strategy_name": "earnings_gapup_breakout",
"summary": "Buys post-earnings gap-up stocks when price holds above VWAP and breaks day-one high.",
"known_pain_points": [
"Late entries after noon show poor expectancy",
"False continuation when opening range volume is weak",
"Stop placement too tight in high ATR names"
]
},
"constraints": {
"execution_constraints": [
"Entries only during regular session",
"No more than one order amendment per position"
],
"risk_constraints": [
"Max 6 concurrent positions",
"risk_per_trade <= 1.0%",
"max single-position size 20%"
],
"data_constraints": [
"No reliance on sub-second data"
]
},
"market_context": {
"summary": "Risk-on tape with concentrated post-earnings leadership in software and semis.",
"regime_tags": [
"risk_on",
"earnings_season"
],
"observations": [
"Leaders often reclaim VWAP within first 30 minutes",
"Continuation quality drops when market breadth deteriorates intraday"
]
},
"performance_summary": {
"win_rate": 0.52,
"profit_factor": 1.18,
"max_drawdown": 0.14
},
"trade_log_summary": {
"common_winner_traits": [
"Relative volume above 2.0 by 10:30",
"First pullback holds above session VWAP"
],
"common_loser_traits": [
"Entry taken after failed second breakout",
"Volume dries below 1.2x by noon"
]
},
"feature_inventory": [
"gap_pct",
"open_to_vwap_distance",
"relative_volume_5m",
"or_high_break_time",
"atr20"
],
"journal_snippets": [
{
"date": "2026-02-10",
"note": "Tried requiring first 15m close above VWAP; improved selectivity but fewer trades.",
"outcome": "mixed"
},
{
"date": "2026-02-15",
"note": "Tried adding tick-level imbalance proxy; not robust across brokers.",
"outcome": "rejected"
},
{
"date": "2026-02-21",
"note": "Using opening range contraction before break looked promising in top decile volume names.",
"outcome": "promising"
}
],
"qualitative_notes": [
"Execution slippage spikes around macro headline releases"
],
"artifacts": [
{
"type": "chartbook",
"path": "reports/earnings_gapup_review_2026w08.pdf"
}
]
}
{
"generated_at_utc": "2026-03-05T00:00:00Z",
"summary": "Prioritize a VWAP-hold continuation entry filter and tighten invalidation logic for weak midday volume.",
"state_assessment": "Evidence quality is moderate-high; setup appears testable under current execution constraints.",
"hypotheses": [
{
"hypothesis_id": "H-001",
"title": "VWAP hold continuation after earnings gap-up",
"thesis": "Gap-up leaders that reclaim and hold VWAP in the first 30 minutes are more likely to continue through day-one high.",
"problem_target": "False continuation entries with weak early-session structure",
"mechanism": "Early institutional participation is proxied by sustained VWAP hold plus above-baseline opening volume.",
"evidence_basis": [
"Winners show >2.0 relative volume by 10:30",
"Losers often lose VWAP before noon"
],
"proposed_rule_changes": [
{
"component": "entry_logic",
"change": "Require 5-minute close above VWAP for 3 consecutive bars before break of day-one high."
},
{
"component": "exit_logic",
"change": "Invalidate if price closes below VWAP with relative_volume_5m < 1.2 before 12:00."
}
],
"expected_impact": "Lower false positives and improved trade quality at cost of fewer entries.",
"key_risks": [
"May reduce trade count in lower-volatility earnings windows",
"VWAP hold may lag in fast trend acceleration"
],
"kill_criteria": [
"If 30-trade sample shows expectancy <= 0 after costs",
"If median adverse excursion worsens by >20% versus baseline"
],
"minimum_viable_experiment": {
"goal": "Test whether VWAP hold precondition improves net expectancy.",
"setup": "Backtest 2019-01 to latest; compare baseline vs VWAP-hold variant.",
"metrics": [
"expectancy_after_costs",
"win_rate",
"max_drawdown"
],
"sample_size": ">= 30 trades",
"duration": "2 weeks"
},
"score_components": {
"evidence_strength": 5,
"mechanism_clarity": 5,
"feasibility": 5,
"expected_payoff": 4,
"novelty": 3,
"test_efficiency": 5
},
"priority_score": 4.65,
"recommendation": "pursue",
"rationale": "Strong fit with observed winner traits and feasible with existing features.",
"confidence": "medium-high",
"assumptions": [
"Relative volume remains predictive in current regime"
],
"dependencies": [
"Reliable intraday VWAP and 5m volume data"
]
}
],
"selected_next_actions": [
"Run A/B backtest baseline vs VWAP-hold variant",
"Review slippage impact around macro event windows"
],
"warnings": [
"Regime shift could reduce transferability"
],
"logging_payload": {
"objective": "Improve consistency of an earnings gap-up continuation setup",
"sources_used": [
"strategy_context",
"trade_log_summary",
"journal_snippets",
"market_context"
],
"hypothesis_ids": [
"H-001"
],
"selected_next_actions": [
"Run A/B backtest baseline vs VWAP-hold variant",
"Review slippage impact around macro event windows"
],
"warnings": [
"Regime shift could reduce transferability"
]
}
}
{
"objective": {
"goal": "Improve quality of breakout entries in earnings season",
"focus_area": "Entry filter and stop placement"
},
"strategy_context": {
"strategy_name": "earnings_breakout_v1",
"summary": "Trade post-earnings breakouts when price confirms above key intraday levels.",
"known_pain_points": [
"False breakouts on weak opening volume",
"Late entries after momentum exhaustion"
]
},
"constraints": {
"execution_constraints": [
"Entries only during regular session",
"No more than one add-on per position"
],
"risk_constraints": [
"Max 5 concurrent positions",
"risk_per_trade <= 1.0%"
],
"data_constraints": [
"No reliance on sub-second data"
]
},
"market_context": {
"summary": "Risk-on regime with concentrated earnings leadership in growth sectors.",
"regime_tags": [
"risk_on",
"earnings_season"
],
"observations": [
"High-relative-volume names tend to hold VWAP in first hour",
"Continuation quality drops when index breadth weakens intraday"
]
},
"trade_log_summary": {
"common_winner_traits": [
"Relative volume above 2.0 by 10:30",
"First pullback holds above intraday VWAP"
],
"common_loser_traits": [
"Breakout fails after second retest",
"Volume contracts sharply before noon"
]
},
"feature_inventory": [
"gap_pct",
"relative_volume_5m",
"open_to_vwap_distance",
"atr20"
],
"journal_snippets": [
{
"date": "2026-02-22",
"note": "Tighter volume filter reduced false starts but lowered trade count.",
"outcome": "mixed"
},
{
"date": "2026-02-26",
"note": "Tick-level imbalance idea was unstable across data providers.",
"outcome": "rejected"
}
]
}
Review each hypothesis card critically.
For each card:
- Point out unsupported claims.
- Flag overfitting and execution realism risks.
- Check whether kill criteria are objective and measurable.
- Suggest a tighter minimum viable experiment when possible.
Return concise critique bullets per hypothesis_id.
Objective
Generate 1-5 high-quality trade hypothesis cards for iterative testing.
Evidence Summary
{{evidence_summary}}
Output Requirements
Return JSON with top-level key hypotheses. Each hypothesis must include all required fields from schemas/hypothesis_card.schema.json. Focus on:
- clear mechanism
- testability with available features
- explicit invalidation and constraints-awareness
You are a trading strategy hypothesis ideation engine.
Produce 1-5 falsifiable hypothesis cards from structured evidence.
Hard requirements:
- Use only evidence provided in the input context.
- Avoid certainty claims and avoid banned phrases.
- Define explicit kill criteria and minimum viable experiment.
- Keep recommendation strictly in {pursue, park, reject}.
- Keep score_components each in [1, 5].
Evidence Quality Guide
Use this lightweight rubric when assembling evidence basis items.
Source Weights (qualitative)
- High: validated trade logs, reproducible backtest outputs, auditable fills
- Medium: structured journal snippets with explicit outcomes, repeated chart observations
- Low: one-off anecdotes, unconstrained intuition, non-reproducible screenshots
Checklist
- Prefer evidence that can be traced to a concrete source.
- Distinguish signal from regime-specific noise.
- Include at least one falsifiable counter-condition in kill criteria.
- Penalize hypotheses that require unavailable features or unsupported data granularity.
Hypothesis Types
Mapping to entry_family
breakout->pivot_breakout(exportable)earnings_drift/gap_continuation->gap_up_continuation(exportable)momentum/pullback/regime_shift->research_only(not exportable in v1)
Notes
edge-finder-candidate/v1supports onlypivot_breakoutandgap_up_continuation.gap_open_scoredis treated as research-only until v2 support is added.
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "HypothesisCard",
"type": "object",
"required": [
"hypothesis_id",
"title",
"thesis",
"problem_target",
"mechanism",
"evidence_basis",
"proposed_rule_changes",
"expected_impact",
"key_risks",
"kill_criteria",
"minimum_viable_experiment",
"score_components",
"priority_score",
"recommendation",
"rationale",
"confidence",
"assumptions",
"dependencies"
],
"properties": {
"hypothesis_id": {"type": "string"},
"title": {"type": "string"},
"thesis": {"type": "string"},
"problem_target": {"type": "string"},
"mechanism": {"type": "string"},
"evidence_basis": {
"type": "array",
"minItems": 1,
"items": {"type": "string"}
},
"proposed_rule_changes": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"required": ["component", "change"],
"properties": {
"component": {"type": "string"},
"change": {"type": "string"}
},
"additionalProperties": true
}
},
"expected_impact": {"type": "string"},
"key_risks": {
"type": "array",
"items": {"type": "string"}
},
"kill_criteria": {
"type": "array",
"minItems": 1,
"items": {"type": "string"}
},
"minimum_viable_experiment": {
"type": "object",
"required": ["goal", "setup", "metrics", "sample_size", "duration"],
"properties": {
"goal": {"type": "string"},
"setup": {"type": "string"},
"metrics": {
"type": "array",
"items": {"type": "string"}
},
"sample_size": {"type": "string"},
"duration": {"type": "string"}
},
"additionalProperties": true
},
"score_components": {
"type": "object",
"required": [
"evidence_strength",
"mechanism_clarity",
"feasibility",
"expected_payoff",
"novelty",
"test_efficiency"
],
"properties": {
"evidence_strength": {"type": "number", "minimum": 1, "maximum": 5},
"mechanism_clarity": {"type": "number", "minimum": 1, "maximum": 5},
"feasibility": {"type": "number", "minimum": 1, "maximum": 5},
"expected_payoff": {"type": "number", "minimum": 1, "maximum": 5},
"novelty": {"type": "number", "minimum": 1, "maximum": 5},
"test_efficiency": {"type": "number", "minimum": 1, "maximum": 5}
},
"additionalProperties": false
},
"priority_score": {"type": "number", "minimum": 1, "maximum": 5},
"recommendation": {
"type": "string",
"enum": ["pursue", "park", "reject"]
},
"rationale": {"type": "string"},
"confidence": {"type": "string"},
"assumptions": {
"type": "array",
"items": {"type": "string"}
},
"dependencies": {
"type": "array",
"items": {"type": "string"}
}
},
"additionalProperties": true
}
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "TradeHypothesisInputBundle",
"type": "object",
"required": [
"objective",
"strategy_context",
"constraints",
"market_context"
],
"properties": {
"objective": {
"type": "object",
"required": ["goal", "focus_area"],
"properties": {
"goal": {"type": "string"},
"focus_area": {"type": "string"},
"horizon": {"type": "string"}
},
"additionalProperties": true
},
"strategy_context": {
"type": "object",
"required": ["strategy_name", "summary"],
"properties": {
"strategy_name": {"type": "string"},
"summary": {"type": "string"},
"known_pain_points": {
"type": "array",
"items": {"type": "string"}
}
},
"additionalProperties": true
},
"constraints": {
"type": "object",
"required": ["execution_constraints", "risk_constraints"],
"properties": {
"execution_constraints": {
"type": "array",
"items": {"type": "string"}
},
"risk_constraints": {
"type": "array",
"items": {"type": "string"}
},
"data_constraints": {
"type": "array",
"items": {"type": "string"}
}
},
"additionalProperties": true
},
"market_context": {
"type": "object",
"required": ["summary"],
"properties": {
"summary": {"type": "string"},
"regime_tags": {
"type": "array",
"items": {"type": "string"}
},
"observations": {
"type": "array",
"items": {"type": "string"}
}
},
"additionalProperties": true
},
"performance_summary": {
"type": ["object", "null"]
},
"trade_log_summary": {
"type": ["object", "null"]
},
"feature_inventory": {
"type": "array",
"items": {"type": "string"}
},
"journal_snippets": {
"type": "array",
"items": {"type": "object"}
},
"qualitative_notes": {
"type": "array",
"items": {"type": "string"}
},
"artifacts": {
"type": "array",
"items": {"type": "object"}
}
},
"additionalProperties": true
}
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "TradeHypothesisOutputBundle",
"type": "object",
"required": [
"generated_at_utc",
"summary",
"state_assessment",
"hypotheses",
"selected_next_actions",
"warnings",
"logging_payload"
],
"properties": {
"generated_at_utc": {"type": "string"},
"summary": {"type": "string"},
"state_assessment": {"type": "string"},
"hypotheses": {
"type": "array",
"minItems": 1,
"maxItems": 5,
"items": {
"$ref": "./hypothesis_card.schema.json"
}
},
"selected_next_actions": {
"type": "array",
"items": {"type": "string"}
},
"warnings": {
"type": "array",
"items": {"type": "string"}
},
"logging_payload": {
"type": "object",
"required": [
"objective",
"sources_used",
"hypothesis_ids",
"selected_next_actions",
"warnings"
],
"properties": {
"objective": {"type": "string"},
"sources_used": {
"type": "array",
"items": {"type": "string"}
},
"hypothesis_ids": {
"type": "array",
"items": {"type": "string"}
},
"selected_next_actions": {
"type": "array",
"items": {"type": "string"}
},
"warnings": {
"type": "array",
"items": {"type": "string"}
}
},
"additionalProperties": true
}
},
"additionalProperties": true
}
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "RawHypothesesInput",
"description": "Reference schema for --hypotheses CLI input.",
"type": "object",
"required": ["hypotheses"],
"properties": {
"hypotheses": {
"type": "array",
"minItems": 1,
"maxItems": 5,
"items": {"type": "object"}
}
},
"additionalProperties": false
}
"""Extract and format evidence signals from normalized input bundle."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
@dataclass
class EvidenceSummary:
"""Canonical evidence summary used by prompt generation."""
primary_issues: list[str]
regime_characteristics: list[str]
winning_patterns: list[str]
losing_patterns: list[str]
available_features: list[str]
rejected_directions: list[str]
execution_constraints: list[str]
def extract_evidence(normalized: dict) -> EvidenceSummary:
"""Extract structured evidence fields from a normalized input bundle."""
strategy_context = normalized.get("strategy_context") if isinstance(normalized, dict) else {}
market_context = normalized.get("market_context") if isinstance(normalized, dict) else {}
trade_log_summary = normalized.get("trade_log_summary") if isinstance(normalized, dict) else {}
constraints = normalized.get("constraints") if isinstance(normalized, dict) else {}
strategy_context = strategy_context if isinstance(strategy_context, dict) else {}
market_context = market_context if isinstance(market_context, dict) else {}
trade_log_summary = trade_log_summary if isinstance(trade_log_summary, dict) else {}
constraints = constraints if isinstance(constraints, dict) else {}
primary_issues = _as_list_of_str(strategy_context.get("known_pain_points"))
regime_characteristics = _as_list_of_str(market_context.get("regime_tags")) + _as_list_of_str(
market_context.get("observations")
)
winning_patterns = _as_list_of_str(trade_log_summary.get("common_winner_traits"))
losing_patterns = _as_list_of_str(trade_log_summary.get("common_loser_traits"))
available_features = _as_list_of_str(normalized.get("feature_inventory"))
rejected_directions: list[str] = []
for snippet in normalized.get("journal_snippets", []):
if not isinstance(snippet, dict):
continue
outcome = str(snippet.get("outcome", "")).strip().lower()
if outcome == "promising":
continue
note = str(snippet.get("note", "")).strip()
if note:
rejected_directions.append(f"[{outcome or 'unknown'}] {note}")
execution_constraints = _as_list_of_str(
constraints.get("execution_constraints")
) + _as_list_of_str(constraints.get("risk_constraints"))
return EvidenceSummary(
primary_issues=primary_issues,
regime_characteristics=regime_characteristics,
winning_patterns=winning_patterns,
losing_patterns=losing_patterns,
available_features=available_features,
rejected_directions=rejected_directions,
execution_constraints=execution_constraints,
)
def format_evidence_for_prompt(summary: EvidenceSummary) -> str:
"""Format evidence summary into a deterministic prompt-friendly markdown block."""
sections = [
("Primary issues", summary.primary_issues),
("Regime characteristics", summary.regime_characteristics),
("Winning patterns", summary.winning_patterns),
("Losing patterns", summary.losing_patterns),
("Available features", summary.available_features),
("Rejected directions", summary.rejected_directions),
("Execution constraints", summary.execution_constraints),
]
lines: list[str] = []
for title, items in sections:
lines.append(f"## {title}")
if not items:
lines.append("- (none)")
else:
lines.extend(f"- {item}" for item in items)
lines.append("")
return "\n".join(lines).strip() + "\n"
def _as_list_of_str(value: Any) -> list[str]:
if not isinstance(value, list):
return []
return [str(item) for item in value if isinstance(item, str) and item.strip()]
"""Output bundle validation, guardrails, and report formatting (Step G)."""
from __future__ import annotations
import re
from difflib import SequenceMatcher
from typing import Any
BANNED_PHRASES = [
"本番投入可能",
"確実に勝てる",
"production ready",
"guaranteed edge",
"sure to win",
"guaranteed profit",
]
_REQUIRED_CARD_FIELDS = (
"hypothesis_id",
"title",
"thesis",
"problem_target",
"mechanism",
"evidence_basis",
"proposed_rule_changes",
"expected_impact",
"key_risks",
"kill_criteria",
"minimum_viable_experiment",
"score_components",
"priority_score",
"recommendation",
"rationale",
"confidence",
"assumptions",
"dependencies",
)
_REQUIRED_MVE_FIELDS = ("goal", "setup", "metrics", "sample_size", "duration")
_ALLOWED_RECOMMENDATIONS = {"pursue", "park", "reject"}
def validate_hypothesis_card(card: dict) -> list[str]:
"""Validate a single hypothesis card payload."""
errors: list[str] = []
if not isinstance(card, dict):
return ["card must be an object"]
for key in _REQUIRED_CARD_FIELDS:
if key not in card:
errors.append(f"{key} is required")
evidence_basis = card.get("evidence_basis")
if not isinstance(evidence_basis, list) or len(evidence_basis) < 1:
errors.append("evidence_basis must be a non-empty list")
kill_criteria = card.get("kill_criteria")
if not isinstance(kill_criteria, list) or len(kill_criteria) < 1:
errors.append("kill_criteria must be a non-empty list")
mve = card.get("minimum_viable_experiment")
if not isinstance(mve, dict):
errors.append("minimum_viable_experiment must be an object")
else:
for key in _REQUIRED_MVE_FIELDS:
value = mve.get(key)
if value is None or (isinstance(value, str) and not value.strip()):
errors.append(f"minimum_viable_experiment.{key} is required")
recommendation = card.get("recommendation")
if recommendation not in _ALLOWED_RECOMMENDATIONS:
errors.append(
f"recommendation must be one of: {', '.join(sorted(_ALLOWED_RECOMMENDATIONS))}"
)
for phrase in _find_banned_phrases(card):
errors.append(f"banned phrase detected in card: {phrase}")
return errors
def check_constraint_violations(card: dict, constraints: dict) -> list[str]:
"""Check whether a hypothesis conflicts with execution/data/risk constraints."""
issues: list[str] = []
if not isinstance(card, dict) or not isinstance(constraints, dict):
return issues
all_text = "\n".join(_extract_string_values(card)).lower()
for constraint in _as_str_list(constraints.get("data_constraints")):
c_lower = constraint.lower()
if "sub-second" in c_lower or "tick" in c_lower:
if any(
token in all_text for token in ("tick", "tick-level", "sub-second", "millisecond")
):
issues.append(
"data constraint violation: "
f"'{constraint}' conflicts with tick/sub-second requirement in hypothesis"
)
for constraint in _as_str_list(constraints.get("risk_constraints")):
c_lower = constraint.lower()
max_pos = _parse_max_positions(c_lower)
if max_pos is not None:
proposed_positions = _find_position_counts(all_text)
for proposed in proposed_positions:
if proposed > max_pos:
issues.append(
f"risk constraint violation: '{constraint}' but proposal references {proposed} positions"
)
max_position_size = _parse_max_position_size_pct(c_lower)
if max_position_size is not None:
for pct in _find_position_size_pcts(all_text):
if pct > max_position_size:
issues.append(
"risk constraint violation: "
f"'{constraint}' but proposal references position size {pct:.1f}%"
)
return issues
def check_duplicate_hypotheses(hypotheses: list[dict]) -> list[str]:
"""Detect duplicate hypotheses by id, title similarity, and mechanism tuple."""
issues: list[str] = []
id_to_first_index: dict[str, int] = {}
norm_title_entries: list[tuple[int, str, str]] = []
mechanism_pairs: dict[tuple[str, str], int] = {}
for idx, card in enumerate(hypotheses):
if not isinstance(card, dict):
issues.append(f"hypotheses[{idx}] must be an object")
continue
hypothesis_id = str(card.get("hypothesis_id", "")).strip()
if hypothesis_id:
if hypothesis_id in id_to_first_index:
first = id_to_first_index[hypothesis_id]
issues.append(
"duplicate hypothesis_id detected: "
f"'{hypothesis_id}' at indices {first} and {idx}"
)
else:
id_to_first_index[hypothesis_id] = idx
title = _normalize_title_for_similarity(str(card.get("title", "")))
if title:
norm_title_entries.append((idx, hypothesis_id or f"idx-{idx}", title))
problem_target = _normalize_text(str(card.get("problem_target", "")))
mechanism = _normalize_text(str(card.get("mechanism", "")))
pair = (problem_target, mechanism)
if pair != ("", ""):
if pair in mechanism_pairs:
first = mechanism_pairs[pair]
issues.append(
f"duplicate problem_target+mechanism detected between indices {first} and {idx}"
)
else:
mechanism_pairs[pair] = idx
for i, (_, id_a, title_a) in enumerate(norm_title_entries):
for _, id_b, title_b in norm_title_entries[i + 1 :]:
ratio = SequenceMatcher(None, title_a, title_b).ratio()
if ratio > 0.8:
issues.append(
f"potential duplicate title pair: {id_a} vs {id_b} (similarity={ratio:.2f})"
)
return issues
def validate_output_bundle(bundle: dict, constraints: dict | None = None) -> list[str]:
"""Validate output bundle shape plus all guardrails."""
errors: list[str] = []
if not isinstance(bundle, dict):
return ["output bundle must be an object"]
hypotheses = bundle.get("hypotheses")
if not isinstance(hypotheses, list):
errors.append("hypotheses must be an array")
hypotheses = []
elif not (1 <= len(hypotheses) <= 5):
errors.append("hypotheses must contain 1-5 cards")
summary = bundle.get("summary", "")
state_assessment = bundle.get("state_assessment", "")
for phrase in _find_banned_phrases({"summary": summary, "state_assessment": state_assessment}):
errors.append(f"banned phrase detected in bundle: {phrase}")
for idx, card in enumerate(hypotheses):
card_errors = validate_hypothesis_card(card)
errors.extend([f"hypotheses[{idx}]: {msg}" for msg in card_errors])
errors.extend(check_duplicate_hypotheses(hypotheses))
if constraints is not None:
for idx, card in enumerate(hypotheses):
issues = check_constraint_violations(card, constraints)
errors.extend([f"hypotheses[{idx}]: {issue}" for issue in issues])
return errors
def build_logging_payload(bundle: dict, normalized_input: dict) -> dict:
"""Build journal-compatible logging payload."""
objective = ""
if isinstance(normalized_input, dict):
objective_obj = normalized_input.get("objective")
if isinstance(objective_obj, dict):
objective = str(objective_obj.get("goal", "")).strip()
hypothesis_ids = [
str(card.get("hypothesis_id"))
for card in bundle.get("hypotheses", [])
if isinstance(card, dict) and card.get("hypothesis_id")
]
candidate_sources = (
"strategy_context",
"constraints",
"market_context",
"performance_summary",
"trade_log_summary",
"feature_inventory",
"journal_snippets",
"qualitative_notes",
"artifacts",
)
sources_used: list[str] = []
for key in candidate_sources:
value = normalized_input.get(key) if isinstance(normalized_input, dict) else None
if value not in (None, [], {}, ""):
sources_used.append(key)
return {
"objective": objective,
"sources_used": sources_used,
"hypothesis_ids": hypothesis_ids,
"selected_next_actions": _as_str_list(bundle.get("selected_next_actions")),
"warnings": _as_str_list(bundle.get("warnings")),
}
def build_markdown_report(bundle: dict) -> str:
"""Build a readable markdown report for hypothesis outcomes."""
lines: list[str] = ["# Trade Hypothesis Report", "", "## Summary"]
lines.append(str(bundle.get("summary", "")) or "(none)")
lines.extend(["", "## State Assessment", str(bundle.get("state_assessment", "")) or "(none)"])
lines.extend(["", "## Hypotheses"])
hypotheses = bundle.get("hypotheses", [])
if not isinstance(hypotheses, list) or not hypotheses:
lines.append("- (none)")
else:
for card in hypotheses:
if not isinstance(card, dict):
continue
hid = card.get("hypothesis_id", "unknown")
title = card.get("title", "untitled")
recommendation = card.get("recommendation", "unknown")
priority = card.get("priority_score", "n/a")
lines.append(f"### {hid}: {title}")
lines.append(f"- Recommendation: {recommendation}")
lines.append(f"- Priority Score: {priority}")
lines.append(f"- Thesis: {card.get('thesis', '')}")
lines.append(f"- Mechanism: {card.get('mechanism', '')}")
lines.append(f"- Expected Impact: {card.get('expected_impact', '')}")
lines.append("")
lines.append("## Risks")
risks: list[str] = []
for card in hypotheses if isinstance(hypotheses, list) else []:
if isinstance(card, dict):
risks.extend(_as_str_list(card.get("key_risks")))
if not risks:
lines.append("- (none)")
else:
for risk in dict.fromkeys(risks):
lines.append(f"- {risk}")
lines.append("")
lines.append("## Next Actions")
actions = _as_str_list(bundle.get("selected_next_actions"))
if not actions:
lines.append("- (none)")
else:
lines.extend(f"- {action}" for action in actions)
return "\n".join(lines).strip() + "\n"
def _find_banned_phrases(obj: Any) -> list[str]:
text = "\n".join(_extract_string_values(obj)).lower()
found: list[str] = []
for phrase in BANNED_PHRASES:
if phrase.lower() in text:
found.append(phrase)
return found
def _extract_string_values(obj: Any) -> list[str]:
values: list[str] = []
if isinstance(obj, str):
return [obj]
if isinstance(obj, dict):
for value in obj.values():
values.extend(_extract_string_values(value))
elif isinstance(obj, list):
for value in obj:
values.extend(_extract_string_values(value))
return values
def _as_str_list(value: Any) -> list[str]:
if not isinstance(value, list):
return []
return [str(item).strip() for item in value if isinstance(item, str) and str(item).strip()]
def _normalize_text(value: str) -> str:
return re.sub(r"\s+", " ", value.strip().lower())
def _normalize_title_for_similarity(value: str) -> str:
base = re.sub(r"[^a-z0-9\s]+", " ", value.lower())
tokens = [token for token in base.split() if token]
stop_words = {
"based",
"entry",
"entries",
"setup",
"signal",
"signals",
"strategy",
"rule",
"rules",
"approach",
}
filtered = [token for token in tokens if token not in stop_words]
if not filtered:
filtered = tokens
return " ".join(filtered)
def _parse_max_positions(text: str) -> int | None:
match = re.search(r"max\s+(\d+)\s+(?:concurrent\s+)?positions?", text)
if not match:
return None
return int(match.group(1))
def _find_position_counts(text: str) -> list[int]:
matches = re.findall(r"(\d+)\s+positions?", text)
return [int(match) for match in matches]
def _parse_max_position_size_pct(text: str) -> float | None:
patterns = (
r"max\s+single-position\s+size\s*(\d+(?:\.\d+)?)%",
r"max\s+position\s+size\s*(\d+(?:\.\d+)?)%",
r"max\s*(\d+(?:\.\d+)?)%\s+position",
)
for pattern in patterns:
match = re.search(pattern, text)
if match:
return float(match.group(1))
return None
def _find_position_size_pcts(text: str) -> list[float]:
matches = re.findall(r"(\d+(?:\.\d+)?)%\s+(?:position|allocation|capital)", text)
return [float(match) for match in matches]
"""Normalization and lightweight schema validation for hypothesis ideation input."""
from __future__ import annotations
from copy import deepcopy
from typing import Any
OPTIONAL_DEFAULTS = {
"performance_summary": None,
"trade_log_summary": None,
"feature_inventory": [],
"journal_snippets": [],
"qualitative_notes": [],
"artifacts": [],
}
RAW_HYPOTHESIS_REQUIRED_FIELDS = (
"hypothesis_id",
"title",
"thesis",
"problem_target",
"mechanism",
"evidence_basis",
"proposed_rule_changes",
"expected_impact",
"key_risks",
"kill_criteria",
"minimum_viable_experiment",
"score_components",
"priority_score",
"recommendation",
"rationale",
"confidence",
"assumptions",
"dependencies",
)
_ALLOWED_RECOMMENDATIONS = {"pursue", "park", "reject"}
_SCORE_KEYS = (
"evidence_strength",
"mechanism_clarity",
"feasibility",
"expected_payoff",
"novelty",
"test_efficiency",
)
def validate_input(data: dict) -> list[str]:
"""Validate required input fields and minimal typing constraints."""
errors: list[str] = []
if not isinstance(data, dict):
return ["input must be a JSON object"]
objective = data.get("objective")
if not isinstance(objective, dict):
errors.append("objective must be an object")
else:
_require_non_empty_string(objective, "goal", errors, "objective.goal")
_require_non_empty_string(objective, "focus_area", errors, "objective.focus_area")
strategy_context = data.get("strategy_context")
if not isinstance(strategy_context, dict):
errors.append("strategy_context must be an object")
else:
_require_non_empty_string(
strategy_context,
"strategy_name",
errors,
"strategy_context.strategy_name",
)
_require_non_empty_string(strategy_context, "summary", errors, "strategy_context.summary")
constraints = data.get("constraints")
if not isinstance(constraints, dict):
errors.append("constraints must be an object")
else:
_require_list(
constraints, "execution_constraints", errors, "constraints.execution_constraints"
)
_require_list(constraints, "risk_constraints", errors, "constraints.risk_constraints")
market_context = data.get("market_context")
if not isinstance(market_context, dict):
errors.append("market_context must be an object")
else:
_require_non_empty_string(market_context, "summary", errors, "market_context.summary")
return errors
def fill_defaults(data: dict) -> dict:
"""Fill optional fields without mutating the caller-owned object."""
normalized = deepcopy(data)
for key, default_value in OPTIONAL_DEFAULTS.items():
if key not in normalized:
normalized[key] = deepcopy(default_value)
return normalized
def normalize(data: dict) -> tuple[dict, list[str]]:
"""Validate input and return a default-filled normalized payload."""
errors = validate_input(data)
normalized = fill_defaults(data if isinstance(data, dict) else {})
return normalized, errors
def validate_raw_hypotheses(payload: dict[str, Any]) -> list[str]:
"""Validate Pass 2 raw hypotheses payload (`--hypotheses`)."""
errors: list[str] = []
if not isinstance(payload, dict):
return ["raw hypotheses payload must be a JSON object"]
extra_keys = sorted(set(payload.keys()) - {"hypotheses"})
if extra_keys:
errors.append(f"raw hypotheses payload has unexpected keys: {', '.join(extra_keys)}")
hypotheses = payload.get("hypotheses")
if hypotheses is None:
errors.append("hypotheses key is required")
return errors
if not isinstance(hypotheses, list):
errors.append("hypotheses must be an array")
return errors
if not (1 <= len(hypotheses) <= 5):
errors.append("hypotheses must contain 1-5 cards")
for idx, card in enumerate(hypotheses):
prefix = f"hypotheses[{idx}]"
if not isinstance(card, dict):
errors.append(f"{prefix} must be an object")
continue
for field in RAW_HYPOTHESIS_REQUIRED_FIELDS:
if field not in card:
errors.append(f"{prefix}.{field} is required")
recommendation = card.get("recommendation")
if recommendation not in _ALLOWED_RECOMMENDATIONS:
errors.append(
f"{prefix}.recommendation must be one of: {', '.join(sorted(_ALLOWED_RECOMMENDATIONS))}"
)
evidence_basis = card.get("evidence_basis")
if not isinstance(evidence_basis, list) or len(evidence_basis) < 1:
errors.append(f"{prefix}.evidence_basis must be a non-empty list")
kill_criteria = card.get("kill_criteria")
if not isinstance(kill_criteria, list) or len(kill_criteria) < 1:
errors.append(f"{prefix}.kill_criteria must be a non-empty list")
minimum_viable_experiment = card.get("minimum_viable_experiment")
if not isinstance(minimum_viable_experiment, dict):
errors.append(f"{prefix}.minimum_viable_experiment must be an object")
else:
for sub_key in ("goal", "setup", "metrics", "sample_size", "duration"):
if sub_key not in minimum_viable_experiment:
errors.append(f"{prefix}.minimum_viable_experiment.{sub_key} is required")
score_components = card.get("score_components")
if not isinstance(score_components, dict):
errors.append(f"{prefix}.score_components must be an object")
continue
for score_key in _SCORE_KEYS:
if score_key not in score_components:
errors.append(f"{prefix}.score_components.{score_key} is required")
continue
score_value = score_components[score_key]
if not _is_number(score_value) or not (1 <= float(score_value) <= 5):
errors.append(f"{prefix}.score_components.{score_key} must be in [1, 5]")
return errors
def _require_non_empty_string(
payload: dict[str, Any],
key: str,
errors: list[str],
label: str,
) -> None:
value = payload.get(key)
if not isinstance(value, str) or not value.strip():
errors.append(f"{label} must be a non-empty string")
def _require_list(payload: dict[str, Any], key: str, errors: list[str], label: str) -> None:
value = payload.get(key)
if not isinstance(value, list):
errors.append(f"{label} must be a list")
def _is_number(value: Any) -> bool:
return isinstance(value, (int, float)) and not isinstance(value, bool)
"""Priority scoring and ranking for hypothesis cards (Step F)."""
from __future__ import annotations
from copy import deepcopy
from typing import Any
SCORING_WEIGHTS = {
"evidence_strength": 0.25,
"mechanism_clarity": 0.20,
"feasibility": 0.20,
"expected_payoff": 0.15,
"novelty": 0.10,
"test_efficiency": 0.10,
}
REQUIRED_SCORE_KEYS = frozenset(SCORING_WEIGHTS.keys())
def validate_score_components(scores: dict) -> list[str]:
"""Validate score_components presence, key set, and numeric ranges."""
errors: list[str] = []
if not isinstance(scores, dict):
return ["score_components must be a mapping"]
present = set(scores.keys())
missing = sorted(REQUIRED_SCORE_KEYS - present)
unexpected = sorted(present - REQUIRED_SCORE_KEYS)
if missing:
errors.append(f"score_components missing keys: {', '.join(missing)}")
if unexpected:
errors.append(f"score_components has unexpected keys: {', '.join(unexpected)}")
for key in REQUIRED_SCORE_KEYS.intersection(present):
value = scores[key]
if not _is_number(value) or not (1 <= float(value) <= 5):
errors.append(f"score_components.{key} must be in [1, 5]")
return errors
def compute_priority_score(score_components: dict[str, float]) -> float:
"""Compute weighted average score and round to two decimal places."""
errors = validate_score_components(score_components)
if errors:
raise ValueError("invalid score_components: " + "; ".join(errors))
weighted_sum = sum(float(score_components[k]) * SCORING_WEIGHTS[k] for k in REQUIRED_SCORE_KEYS)
return round(weighted_sum, 2)
def rank_hypotheses(hypotheses: list[dict]) -> list[dict]:
"""Compute priority_score for each hypothesis and return descending order."""
ranked: list[dict] = []
for idx, hypothesis in enumerate(hypotheses):
if not isinstance(hypothesis, dict):
raise ValueError(f"invalid hypothesis at index {idx}: must be an object")
score_components = hypothesis.get("score_components")
errors = validate_score_components(score_components)
if errors:
hypothesis_id = hypothesis.get("hypothesis_id", f"idx-{idx}")
raise ValueError(f"invalid score_components for {hypothesis_id}: " + "; ".join(errors))
row = deepcopy(hypothesis)
row["priority_score"] = compute_priority_score(score_components)
ranked.append(row)
ranked.sort(key=lambda x: x["priority_score"], reverse=True)
return ranked
def _is_number(value: Any) -> bool:
return isinstance(value, (int, float)) and not isinstance(value, bool)
"""Step H: export pursue hypotheses into edge-finder candidate artifacts."""
from __future__ import annotations
import importlib.util
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import yaml
INTERFACE_VERSION = "edge-finder-candidate/v1"
EXPORTABLE_ENTRY_TYPES = {"pivot_breakout", "gap_up_continuation"}
ENTRY_FAMILY_KEYWORDS = {
"pivot_breakout": ["breakout", "pivot", "vcp", "contraction", "volatility contraction"],
"gap_up_continuation": ["gap", "earnings gap", "continuation", "gap-up", "vwap hold"],
}
CANDIDATE_CONTRACT_PATH = (
Path(__file__).resolve().parents[3]
/ "edge-candidate-agent"
/ "scripts"
/ "candidate_contract.py"
)
DEFAULT_UNIVERSE = {
"type": "us_equities",
"index": "sp500",
"filters": ["avg_volume > 500_000", "price > 10"],
}
DEFAULT_EXIT = {
"stop_loss": "7% below entry",
"trailing_stop": "below 21-day EMA or 10-day low",
"take_profit": "risk_reward_3x",
"stop_loss_pct": 0.07,
"take_profit_rr": 3.0,
"breakeven_at_rr": 1.0,
}
DEFAULT_RISK = {
"position_sizing": "fixed_risk",
"risk_per_trade": 0.01,
"max_positions": 5,
"max_sector_exposure": 0.30,
}
DEFAULT_COST_MODEL = {
"commission_per_share": 0.00,
"slippage_bps": 5,
}
DEFAULT_PROMOTION_GATES = {
"min_trades": 200,
"max_drawdown": 0.15,
"sharpe": 1.0,
"profit_factor": 1.2,
}
DEFAULT_ENTRY_BY_FAMILY = {
"pivot_breakout": {
"conditions": [
"vcp_pattern_detected",
"breakout_above_pivot_point",
"volume > 1.5 * avg_volume_50",
],
"trend_filter": ["price > sma_200", "price > sma_50", "sma_50 > sma_200"],
},
"gap_up_continuation": {
"conditions": [
"gap_up_detected",
"vwap_hold_in_first_30m",
"breakout_above_gap_day_high",
],
"trend_filter": ["price > sma_200", "price > sma_50", "sma_50 > sma_200"],
},
}
DEFAULT_VCP_DETECTION = {
"min_contractions": 2,
"contraction_ratio": 0.75,
"lookback_window": 120,
"volume_decline": True,
"breakout_volume_ratio": 1.5,
}
DEFAULT_GAP_DETECTION = {
"min_gap_pct": 0.06,
"volume_ratio": 2.0,
"avg_volume_window": 50,
"max_entry_days": 5,
"max_stop_pct": 0.10,
}
def infer_entry_family(hypothesis: dict) -> str | None:
"""Infer exportable entry family from title/thesis/proposed_rule_changes keywords."""
if not isinstance(hypothesis, dict):
return None
text_segments = [
str(hypothesis.get("title", "")),
str(hypothesis.get("thesis", "")),
]
proposed_rule_changes = hypothesis.get("proposed_rule_changes")
if isinstance(proposed_rule_changes, list):
for row in proposed_rule_changes:
if isinstance(row, dict):
text_segments.append(str(row.get("change", "")))
joined = "\n".join(text_segments).lower()
if "gap_open_scored" in joined or "gap open scored" in joined:
return None
family_scores: dict[str, int] = {}
for family, keywords in ENTRY_FAMILY_KEYWORDS.items():
score = sum(1 for kw in keywords if kw.lower() in joined)
if score > 0:
family_scores[family] = score
if not family_scores:
return None
entry_logic_text = ""
if isinstance(proposed_rule_changes, list):
for row in proposed_rule_changes:
if isinstance(row, dict) and str(row.get("component", "")).strip() == "entry_logic":
entry_logic_text = str(row.get("change", "")).lower()
break
if len(family_scores) > 1 and entry_logic_text:
entry_logic_matches = [
family
for family, keywords in ENTRY_FAMILY_KEYWORDS.items()
if any(kw.lower() in entry_logic_text for kw in keywords)
]
if len(entry_logic_matches) == 1:
return entry_logic_matches[0]
if len(entry_logic_matches) > 1:
return max(entry_logic_matches, key=lambda family: family_scores.get(family, 0))
best_score = max(family_scores.values())
contenders = [family for family, score in family_scores.items() if score == best_score]
if len(contenders) == 1:
return contenders[0]
return sorted(contenders)[0]
def can_export(hypothesis: dict) -> bool:
"""Return True if hypothesis can be exported under v1 contract constraints."""
if not isinstance(hypothesis, dict):
return False
return (
hypothesis.get("recommendation") == "pursue" and infer_entry_family(hypothesis) is not None
)
def build_strategy_yaml(hypothesis: dict, candidate_id: str) -> dict:
"""Build strategy.yaml payload for edge-finder-candidate/v1."""
entry_family = infer_entry_family(hypothesis)
if entry_family not in EXPORTABLE_ENTRY_TYPES:
raise ValueError("hypothesis is not exportable under edge-finder-candidate/v1")
default_entry = DEFAULT_ENTRY_BY_FAMILY[entry_family]
entry_conditions = _extract_entry_conditions(hypothesis)
if not entry_conditions:
entry_conditions = default_entry["conditions"]
spec: dict[str, Any] = {
"id": candidate_id,
"name": str(hypothesis.get("title", candidate_id)).strip() or candidate_id,
"description": str(hypothesis.get("thesis", "")).strip(),
"universe": dict(DEFAULT_UNIVERSE),
"signals": {
"entry": {
"type": entry_family,
"conditions": entry_conditions,
"trend_filter": list(default_entry["trend_filter"]),
},
"exit": dict(DEFAULT_EXIT),
},
"risk": dict(DEFAULT_RISK),
"cost_model": dict(DEFAULT_COST_MODEL),
"validation": {
"method": "full_sample",
"oos_ratio": None,
},
"promotion_gates": dict(DEFAULT_PROMOTION_GATES),
}
if entry_family == "pivot_breakout":
spec["vcp_detection"] = dict(DEFAULT_VCP_DETECTION)
elif entry_family == "gap_up_continuation":
spec["gap_up_detection"] = dict(DEFAULT_GAP_DETECTION)
return spec
def validate_strategy_yaml(spec: dict, candidate_id: str) -> list[str]:
"""Validate generated strategy yaml using edge-candidate-agent contract validator."""
validate_interface_contract = _load_candidate_contract_validator()
return validate_interface_contract(spec, candidate_id, stage="phase1")
def build_metadata_json(hypothesis: dict, candidate_id: str) -> dict:
"""Build metadata.json payload with provenance context."""
entry_family = infer_entry_family(hypothesis)
thesis = str(hypothesis.get("thesis", "")).strip()
thesis_summary = thesis if len(thesis) <= 180 else thesis[:177] + "..."
return {
"interface_version": INTERFACE_VERSION,
"candidate_id": candidate_id,
"generated_at_utc": datetime.now(timezone.utc)
.replace(microsecond=0)
.isoformat()
.replace("+00:00", "Z"),
"generator": {
"name": "trade-hypothesis-ideator",
"version": "0.1.0",
},
"research_context": {
"hypothesis_id": str(hypothesis.get("hypothesis_id", "")).strip(),
"entry_family": entry_family,
"thesis": thesis_summary,
},
}
def export_candidate(hypothesis: dict, output_dir: Path, dry_run: bool = False) -> Path | None:
"""Build, validate, and write candidate artifacts from one hypothesis card."""
if not can_export(hypothesis):
return None
candidate_id = _to_candidate_id(hypothesis)
spec = build_strategy_yaml(hypothesis, candidate_id)
errors = validate_strategy_yaml(spec, candidate_id)
if errors:
raise ValueError("strategy export validation failed: " + "; ".join(errors))
metadata = build_metadata_json(hypothesis, candidate_id)
if dry_run:
return None
candidate_dir = output_dir / candidate_id
candidate_dir.mkdir(parents=True, exist_ok=True)
strategy_path = candidate_dir / "strategy.yaml"
metadata_path = candidate_dir / "metadata.json"
strategy_path.write_text(yaml.safe_dump(spec, sort_keys=False, allow_unicode=False))
metadata_path.write_text(json.dumps(metadata, indent=2, ensure_ascii=True) + "\n")
return candidate_dir
def _load_candidate_contract_validator():
if not CANDIDATE_CONTRACT_PATH.exists():
raise RuntimeError(
f"candidate_contract.py not found at expected path: {CANDIDATE_CONTRACT_PATH}"
)
module_name = "edge_candidate_contract_dynamic"
spec = importlib.util.spec_from_file_location(module_name, CANDIDATE_CONTRACT_PATH)
if spec is None or spec.loader is None:
raise RuntimeError(f"failed to build import spec for: {CANDIDATE_CONTRACT_PATH}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
validate_interface_contract = getattr(module, "validate_interface_contract", None)
if validate_interface_contract is None:
raise RuntimeError("candidate_contract.py missing validate_interface_contract()")
return validate_interface_contract
def _extract_entry_conditions(hypothesis: dict) -> list[str]:
changes = hypothesis.get("proposed_rule_changes")
if not isinstance(changes, list):
return []
conditions: list[str] = []
for row in changes:
if not isinstance(row, dict):
continue
if str(row.get("component", "")).strip() != "entry_logic":
continue
change = str(row.get("change", "")).strip()
if change:
conditions.append(change)
return conditions
def _to_candidate_id(hypothesis: dict) -> str:
raw_id = str(hypothesis.get("hypothesis_id", "hypothesis")).strip().lower() or "hypothesis"
norm = re.sub(r"[^a-z0-9]+", "_", raw_id).strip("_")
return f"edge_{norm}_v1"
#!/usr/bin/env python3
"""Run trade-hypothesis-ideator pipeline in two-pass mode."""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import asdict
from datetime import datetime, timezone
from pathlib import Path
from pipeline.evidence_extractor import extract_evidence, format_evidence_for_prompt
from pipeline.format_output import (
build_logging_payload,
build_markdown_report,
validate_output_bundle,
)
from pipeline.normalize import normalize, validate_raw_hypotheses
from pipeline.ranking import rank_hypotheses
from pipeline.strategy_exporter import can_export, export_candidate
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"trade-hypothesis-ideator: pass1 (evidence summary) or pass2 "
"(rank/format/export hypotheses)"
)
)
parser.add_argument("--input", required=True, help="Path to input bundle JSON")
parser.add_argument(
"--hypotheses",
default=None,
help="Path to raw hypotheses JSON (pass2). If omitted, pass1 runs.",
)
parser.add_argument("--output-dir", required=True, help="Directory for generated artifacts")
parser.add_argument(
"--export-strategies",
action="store_true",
help="Export pursue hypotheses into strategy.yaml + metadata.json",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
input_path = Path(args.input).resolve()
output_dir = Path(args.output_dir).resolve()
output_dir.mkdir(parents=True, exist_ok=True)
try:
input_payload = json.loads(input_path.read_text())
except FileNotFoundError:
_print_err(f"input file not found: {input_path}")
return 1
except json.JSONDecodeError as exc:
_print_err(f"failed to parse input JSON: {exc}")
return 1
normalized, input_errors = normalize(input_payload)
if input_errors:
_print_errors("input validation failed", input_errors)
return 1
if not args.hypotheses:
return _run_pass1(normalized, output_dir)
return _run_pass2(
normalized, Path(args.hypotheses).resolve(), output_dir, args.export_strategies
)
def _run_pass1(normalized: dict, output_dir: Path) -> int:
evidence = extract_evidence(normalized)
prompt_ready = format_evidence_for_prompt(evidence)
evidence_payload = {
"generated_at_utc": _utc_now(),
"objective": normalized.get("objective", {}),
"evidence_summary": asdict(evidence),
"prompt_ready_text": prompt_ready,
}
out_path = output_dir / "evidence_summary.json"
out_path.write_text(json.dumps(evidence_payload, indent=2, ensure_ascii=True) + "\n")
print(f"[OK] pass1 completed: {out_path}")
return 0
def _run_pass2(
normalized: dict,
hypotheses_path: Path,
output_dir: Path,
export_strategies: bool,
) -> int:
try:
raw_payload = json.loads(hypotheses_path.read_text())
except FileNotFoundError:
_print_err(f"hypotheses file not found: {hypotheses_path}")
return 1
except json.JSONDecodeError as exc:
_print_err(f"failed to parse hypotheses JSON: {exc}")
return 1
raw_errors = validate_raw_hypotheses(raw_payload)
if raw_errors:
_print_errors("raw hypotheses validation failed", raw_errors)
return 1
try:
ranked_hypotheses = rank_hypotheses(raw_payload["hypotheses"])
except ValueError as exc:
_print_err(str(exc))
return 1
bundle = {
"generated_at_utc": _utc_now(),
"summary": _build_summary(ranked_hypotheses),
"state_assessment": "Ranked hypotheses validated against guardrails.",
"hypotheses": ranked_hypotheses,
"selected_next_actions": _build_next_actions(ranked_hypotheses),
"warnings": [],
}
output_errors = validate_output_bundle(bundle, constraints=normalized.get("constraints"))
if output_errors:
_print_errors("output validation failed", output_errors)
return 1
bundle["logging_payload"] = build_logging_payload(bundle, normalized)
bundle_path = output_dir / "output_bundle.json"
report_path = output_dir / "hypothesis_report.md"
bundle_path.write_text(json.dumps(bundle, indent=2, ensure_ascii=True) + "\n")
report_path.write_text(build_markdown_report(bundle))
print(f"[OK] pass2 bundle: {bundle_path}")
print(f"[OK] pass2 report: {report_path}")
if export_strategies:
exported = _export_strategies(ranked_hypotheses, output_dir)
print(f"[OK] exported strategies: {len(exported)}")
return 0
def _export_strategies(hypotheses: list[dict], output_dir: Path) -> list[str]:
strategies_dir = output_dir / "strategies"
exported_ids: list[str] = []
for card in hypotheses:
if not can_export(card):
continue
candidate_dir = export_candidate(card, strategies_dir, dry_run=False)
if candidate_dir is not None:
exported_ids.append(candidate_dir.name)
manifest = {
"generated_at_utc": _utc_now(),
"count": len(exported_ids),
"candidate_ids": exported_ids,
}
manifest_path = output_dir / "strategy_exports.json"
manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=True) + "\n")
return exported_ids
def _build_summary(hypotheses: list[dict]) -> str:
if not hypotheses:
return "No hypotheses were ranked."
top = hypotheses[0]
return (
"Top hypothesis: "
f"{top.get('hypothesis_id', 'unknown')} ({top.get('title', 'untitled')}) "
f"with priority score {top.get('priority_score', 'n/a')}."
)
def _build_next_actions(hypotheses: list[dict]) -> list[str]:
actions: list[str] = []
for card in hypotheses:
if card.get("recommendation") != "pursue":
continue
mve = card.get("minimum_viable_experiment")
if isinstance(mve, dict):
goal = str(mve.get("goal", "")).strip()
if goal:
actions.append(f"Run MVE: {goal}")
if not actions:
actions.append("Review ranked hypotheses and define one testable MVE.")
return actions
def _print_err(message: str) -> None:
print(f"[ERROR] {message}", file=sys.stderr)
def _print_errors(prefix: str, errors: list[str]) -> None:
print(f"[ERROR] {prefix}", file=sys.stderr)
for err in errors:
print(f" - {err}", file=sys.stderr)
def _utc_now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
if __name__ == "__main__":
raise SystemExit(main())
"""Shared fixtures for trade-hypothesis-ideator tests."""
from __future__ import annotations
import json
import sys
from copy import deepcopy
from pathlib import Path
import pytest
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
SKILL_DIR = SCRIPTS_DIR.parent
EXAMPLES_DIR = SKILL_DIR / "examples"
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
@pytest.fixture()
def example_input() -> dict:
"""Load example input bundle."""
return json.loads((EXAMPLES_DIR / "example_input.json").read_text())
@pytest.fixture()
def example_output() -> dict:
"""Load example output bundle."""
return json.loads((EXAMPLES_DIR / "example_output.json").read_text())
@pytest.fixture()
def base_hypothesis_card(example_output: dict) -> dict:
"""Provide a mutable baseline hypothesis card for tests."""
return deepcopy(example_output["hypotheses"][0])
@pytest.fixture()
def raw_hypotheses_payload(example_output: dict) -> dict:
"""Build --hypotheses payload shape from example output."""
return {"hypotheses": deepcopy(example_output["hypotheses"])}
"""Tests for evidence extraction helpers (Step B)."""
from __future__ import annotations
from copy import deepcopy
from pipeline.evidence_extractor import extract_evidence, format_evidence_for_prompt
def test_extract_primary_issues_from_known_pain_points(example_input: dict) -> None:
summary = extract_evidence(example_input)
assert summary.primary_issues == example_input["strategy_context"]["known_pain_points"]
def test_extract_regime_characteristics_from_tags_and_observations(example_input: dict) -> None:
summary = extract_evidence(example_input)
expected = (
example_input["market_context"]["regime_tags"]
+ example_input["market_context"]["observations"]
)
assert summary.regime_characteristics == expected
def test_extract_winner_and_loser_traits_separately(example_input: dict) -> None:
summary = extract_evidence(example_input)
assert summary.winning_patterns == example_input["trade_log_summary"]["common_winner_traits"]
assert summary.losing_patterns == example_input["trade_log_summary"]["common_loser_traits"]
def test_extract_rejected_directions_from_non_promising_journal_items(example_input: dict) -> None:
summary = extract_evidence(example_input)
assert len(summary.rejected_directions) == 2
assert any("mixed" in item for item in summary.rejected_directions)
assert any("rejected" in item for item in summary.rejected_directions)
def test_extract_available_features_preserves_input_order(example_input: dict) -> None:
summary = extract_evidence(example_input)
assert summary.available_features == example_input["feature_inventory"]
def test_extract_evidence_handles_missing_optional_fields(example_input: dict) -> None:
payload = deepcopy(example_input)
payload["strategy_context"].pop("known_pain_points", None)
payload["market_context"].pop("regime_tags", None)
payload["market_context"].pop("observations", None)
payload["trade_log_summary"] = None
payload["feature_inventory"] = []
payload["journal_snippets"] = []
summary = extract_evidence(payload)
assert summary.primary_issues == []
assert summary.regime_characteristics == []
assert summary.winning_patterns == []
assert summary.losing_patterns == []
assert summary.available_features == []
assert summary.rejected_directions == []
def test_format_evidence_for_prompt_contains_named_sections(example_input: dict) -> None:
summary = extract_evidence(example_input)
text = format_evidence_for_prompt(summary)
assert "Primary issues" in text
assert "Regime characteristics" in text
assert "Winning patterns" in text
assert "Losing patterns" in text
assert "Execution constraints" in text
"""Tests for output formatting and guardrail validation (Step G)."""
from __future__ import annotations
from copy import deepcopy
from pipeline.format_output import (
build_logging_payload,
build_markdown_report,
check_duplicate_hypotheses,
validate_hypothesis_card,
validate_output_bundle,
)
def _bundle_from_card(card: dict) -> dict:
return {
"generated_at_utc": "2026-03-05T00:00:00Z",
"summary": "test summary",
"state_assessment": "test assessment",
"hypotheses": [card],
"selected_next_actions": ["run backtest"],
"warnings": [],
}
def test_validate_hypothesis_card_valid_card_passes(base_hypothesis_card: dict) -> None:
assert validate_hypothesis_card(base_hypothesis_card) == []
def test_validate_hypothesis_card_empty_evidence_basis_fails(base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["evidence_basis"] = []
errors = validate_hypothesis_card(card)
assert errors
assert any("evidence_basis" in err for err in errors)
def test_validate_hypothesis_card_empty_kill_criteria_fails(base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["kill_criteria"] = []
errors = validate_hypothesis_card(card)
assert errors
assert any("kill_criteria" in err for err in errors)
def test_validate_hypothesis_card_missing_mve_goal_fails(base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["minimum_viable_experiment"].pop("goal")
errors = validate_hypothesis_card(card)
assert errors
assert any("minimum_viable_experiment.goal" in err for err in errors)
def test_validate_output_bundle_rejects_more_than_five_hypotheses(
base_hypothesis_card: dict,
) -> None:
bundle = _bundle_from_card(base_hypothesis_card)
bundle["hypotheses"] = [deepcopy(base_hypothesis_card) for _ in range(6)]
for idx, card in enumerate(bundle["hypotheses"]):
card["hypothesis_id"] = f"H-{idx + 1:03d}"
errors = validate_output_bundle(bundle)
assert errors
assert any("1-5" in err for err in errors)
def test_banned_phrase_detection_japanese(base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["thesis"] = "この手法は確実に勝てる。"
errors = validate_hypothesis_card(card)
assert errors
assert any("確実に勝てる" in err for err in errors)
def test_banned_phrase_detection_english(base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["mechanism"] = "This setup is production ready in all regimes."
errors = validate_hypothesis_card(card)
assert errors
assert any("production ready" in err.lower() for err in errors)
def test_duplicate_hypothesis_id_is_detected(base_hypothesis_card: dict) -> None:
card1 = deepcopy(base_hypothesis_card)
card2 = deepcopy(base_hypothesis_card)
issues = check_duplicate_hypotheses([card1, card2])
assert issues
assert any("hypothesis_id" in issue for issue in issues)
def test_similar_titles_over_threshold_are_reported(base_hypothesis_card: dict) -> None:
card1 = deepcopy(base_hypothesis_card)
card1["hypothesis_id"] = "H-001"
card1["title"] = "VWAP hold continuation"
card2 = deepcopy(base_hypothesis_card)
card2["hypothesis_id"] = "H-002"
card2["title"] = "VWAP-based continuation entry"
issues = check_duplicate_hypotheses([card1, card2])
assert issues
assert any("title" in issue.lower() for issue in issues)
def test_build_logging_payload_contains_required_fields(
example_output: dict, example_input: dict
) -> None:
payload = build_logging_payload(example_output, example_input)
for key in (
"objective",
"sources_used",
"hypothesis_ids",
"selected_next_actions",
"warnings",
):
assert key in payload
def test_build_markdown_report_contains_sections(example_output: dict) -> None:
report = build_markdown_report(example_output)
lowered = report.lower()
assert "## summary" in lowered
assert "## hypotheses" in lowered
assert "## risks" in lowered
"""Cross-cutting guardrail tests for output validation."""
from __future__ import annotations
from copy import deepcopy
from pipeline.format_output import (
check_constraint_violations,
check_duplicate_hypotheses,
validate_output_bundle,
)
def _bundle(card: dict) -> dict:
return {
"generated_at_utc": "2026-03-05T00:00:00Z",
"summary": "guardrail summary",
"state_assessment": "guardrail assessment",
"hypotheses": [card],
"selected_next_actions": ["run test"],
"warnings": [],
}
class TestFieldCompleteness:
"""evidence_basis, kill_criteria, MVE, and count checks."""
def test_empty_evidence_basis_fails(self, base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["evidence_basis"] = []
assert validate_output_bundle(_bundle(card))
def test_empty_kill_criteria_fails(self, base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["kill_criteria"] = []
assert validate_output_bundle(_bundle(card))
def test_empty_mve_fails(self, base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["minimum_viable_experiment"] = {}
errors = validate_output_bundle(_bundle(card))
assert any("minimum_viable_experiment" in err for err in errors)
def test_hypotheses_exceeds_five_fails(self, base_hypothesis_card: dict) -> None:
bundle = _bundle(base_hypothesis_card)
bundle["hypotheses"] = [deepcopy(base_hypothesis_card) for _ in range(6)]
for idx, card in enumerate(bundle["hypotheses"]):
card["hypothesis_id"] = f"H-{idx + 1:03d}"
assert validate_output_bundle(bundle)
class TestBannedPhrases:
"""Banned phrase detection."""
def test_banned_phrase_japanese(self, base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["thesis"] = "このルールは本番投入可能で確実に勝てる。"
errors = validate_output_bundle(_bundle(card))
assert any("本番投入可能" in err for err in errors)
def test_banned_phrase_english(self, base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["mechanism"] = "This is production ready with guaranteed edge."
errors = validate_output_bundle(_bundle(card))
assert any("production ready" in err.lower() for err in errors)
class TestConstraintViolation:
"""Constraint violation detection via check_constraint_violations."""
def test_data_constraint_violation(self, base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["proposed_rule_changes"][0]["change"] = "Use tick-level data to detect order flow."
constraints = {"data_constraints": ["No reliance on sub-second data"]}
issues = check_constraint_violations(card, constraints)
assert issues
assert any("sub-second" in issue.lower() for issue in issues)
def test_risk_constraint_violation(self, base_hypothesis_card: dict) -> None:
card = deepcopy(base_hypothesis_card)
card["proposed_rule_changes"][0]["change"] = (
"Increase to 10 positions during earnings week."
)
constraints = {"risk_constraints": ["Max 6 concurrent positions"]}
issues = check_constraint_violations(card, constraints)
assert issues
assert any("max 6" in issue.lower() or "10" in issue for issue in issues)
def test_no_violation_when_constraints_respected(self, base_hypothesis_card: dict) -> None:
constraints = {
"data_constraints": ["No reliance on sub-second data"],
"risk_constraints": ["Max 6 concurrent positions"],
}
assert check_constraint_violations(base_hypothesis_card, constraints) == []
class TestDuplicateDetection:
"""Duplicate hypothesis detection checks."""
def test_identical_hypothesis_id(self, base_hypothesis_card: dict) -> None:
card1 = deepcopy(base_hypothesis_card)
card2 = deepcopy(base_hypothesis_card)
issues = check_duplicate_hypotheses([card1, card2])
assert any("hypothesis_id" in issue for issue in issues)
def test_similar_titles_detected(self, base_hypothesis_card: dict) -> None:
card1 = deepcopy(base_hypothesis_card)
card1["hypothesis_id"] = "H-001"
card1["title"] = "VWAP hold continuation"
card2 = deepcopy(base_hypothesis_card)
card2["hypothesis_id"] = "H-002"
card2["title"] = "VWAP-based continuation entry"
issues = check_duplicate_hypotheses([card1, card2])
assert any("title" in issue.lower() for issue in issues)
def test_distinct_hypotheses_pass(self, base_hypothesis_card: dict) -> None:
card1 = deepcopy(base_hypothesis_card)
card1["hypothesis_id"] = "H-001"
card1["title"] = "VWAP hold continuation"
card2 = deepcopy(base_hypothesis_card)
card2["hypothesis_id"] = "H-002"
card2["title"] = "Pivot breakout after contraction"
card2["problem_target"] = "False breakouts in low breadth"
card2["mechanism"] = "Contraction + volume expansion indicates new demand."
assert check_duplicate_hypotheses([card1, card2]) == []
class TestGoldenPath:
def test_valid_output_passes_all_guardrails(
self, example_output: dict, example_input: dict
) -> None:
errors = validate_output_bundle(example_output, constraints=example_input["constraints"])
assert errors == []
"""Tests for Step A normalization helpers."""
from __future__ import annotations
from copy import deepcopy
from pipeline.normalize import fill_defaults, normalize, validate_input, validate_raw_hypotheses
def test_validate_input_accepts_example_input(example_input: dict) -> None:
errors = validate_input(example_input)
assert errors == []
def test_validate_input_missing_objective_reports_error(example_input: dict) -> None:
payload = deepcopy(example_input)
payload.pop("objective")
errors = validate_input(payload)
assert errors
assert any("objective" in err for err in errors)
def test_validate_input_missing_strategy_name_reports_error(example_input: dict) -> None:
payload = deepcopy(example_input)
payload["strategy_context"].pop("strategy_name")
errors = validate_input(payload)
assert errors
assert any("strategy_context.strategy_name" in err for err in errors)
def test_validate_input_missing_execution_constraints_reports_error(example_input: dict) -> None:
payload = deepcopy(example_input)
payload["constraints"].pop("execution_constraints")
errors = validate_input(payload)
assert errors
assert any("constraints.execution_constraints" in err for err in errors)
def test_validate_input_missing_market_summary_reports_error(example_input: dict) -> None:
payload = deepcopy(example_input)
payload["market_context"].pop("summary")
errors = validate_input(payload)
assert errors
assert any("market_context.summary" in err for err in errors)
def test_validate_input_invalid_objective_type_reports_error(example_input: dict) -> None:
payload = deepcopy(example_input)
payload["objective"] = "improve edge"
errors = validate_input(payload)
assert errors
assert any("objective must be an object" in err for err in errors)
def test_fill_defaults_sets_optional_fields(example_input: dict) -> None:
payload = deepcopy(example_input)
payload.pop("performance_summary", None)
payload.pop("trade_log_summary", None)
payload.pop("feature_inventory", None)
payload.pop("journal_snippets", None)
payload.pop("qualitative_notes", None)
payload.pop("artifacts", None)
normalized = fill_defaults(payload)
assert normalized["performance_summary"] is None
assert normalized["trade_log_summary"] is None
assert normalized["feature_inventory"] == []
assert normalized["journal_snippets"] == []
assert normalized["qualitative_notes"] == []
assert normalized["artifacts"] == []
def test_fill_defaults_does_not_mutate_input(example_input: dict) -> None:
payload = deepcopy(example_input)
payload.pop("feature_inventory")
copied_before = deepcopy(payload)
_ = fill_defaults(payload)
assert payload == copied_before
def test_normalize_returns_filled_payload_and_no_errors(example_input: dict) -> None:
payload = deepcopy(example_input)
payload.pop("qualitative_notes")
normalized, errors = normalize(payload)
assert errors == []
assert normalized["qualitative_notes"] == []
def test_normalize_returns_errors_for_invalid_input(example_input: dict) -> None:
payload = deepcopy(example_input)
payload["market_context"] = "risk_on"
normalized, errors = normalize(payload)
assert isinstance(normalized, dict)
assert errors
assert any("market_context" in err for err in errors)
def test_validate_raw_hypotheses_accepts_valid_payload(raw_hypotheses_payload: dict) -> None:
errors = validate_raw_hypotheses(raw_hypotheses_payload)
assert errors == []
def test_validate_raw_hypotheses_requires_hypotheses_key(raw_hypotheses_payload: dict) -> None:
payload = deepcopy(raw_hypotheses_payload)
payload.pop("hypotheses")
errors = validate_raw_hypotheses(payload)
assert errors
assert any("hypotheses" in err for err in errors)
def test_validate_raw_hypotheses_rejects_empty_array(raw_hypotheses_payload: dict) -> None:
payload = deepcopy(raw_hypotheses_payload)
payload["hypotheses"] = []
errors = validate_raw_hypotheses(payload)
assert errors
assert any("1-5" in err for err in errors)
def test_validate_raw_hypotheses_rejects_more_than_five(raw_hypotheses_payload: dict) -> None:
payload = deepcopy(raw_hypotheses_payload)
payload["hypotheses"] = payload["hypotheses"] * 6
errors = validate_raw_hypotheses(payload)
assert errors
assert any("1-5" in err for err in errors)
def test_validate_raw_hypotheses_requires_hypothesis_id(raw_hypotheses_payload: dict) -> None:
payload = deepcopy(raw_hypotheses_payload)
payload["hypotheses"][0].pop("hypothesis_id")
errors = validate_raw_hypotheses(payload)
assert errors
assert any("hypothesis_id" in err for err in errors)
def test_validate_raw_hypotheses_recommendation_enum(raw_hypotheses_payload: dict) -> None:
payload = deepcopy(raw_hypotheses_payload)
payload["hypotheses"][0]["recommendation"] = "buy"
errors = validate_raw_hypotheses(payload)
assert errors
assert any("recommendation" in err for err in errors)
def test_validate_raw_hypotheses_score_component_range(raw_hypotheses_payload: dict) -> None:
payload = deepcopy(raw_hypotheses_payload)
payload["hypotheses"][0]["score_components"]["novelty"] = 6
errors = validate_raw_hypotheses(payload)
assert errors
assert any("score_components.novelty" in err for err in errors)
"""Tests for ranking helpers (Step F)."""
from __future__ import annotations
from copy import deepcopy
import pytest
from pipeline.ranking import (
REQUIRED_SCORE_KEYS,
SCORING_WEIGHTS,
compute_priority_score,
rank_hypotheses,
validate_score_components,
)
def test_scoring_weights_sum_to_one() -> None:
assert sum(SCORING_WEIGHTS.values()) == pytest.approx(1.0)
def test_compute_priority_score_all_five() -> None:
scores = {key: 5 for key in REQUIRED_SCORE_KEYS}
assert compute_priority_score(scores) == 5.0
def test_compute_priority_score_all_one() -> None:
scores = {key: 1 for key in REQUIRED_SCORE_KEYS}
assert compute_priority_score(scores) == 1.0
def test_compute_priority_score_example_h001_components() -> None:
scores = {
"evidence_strength": 5,
"mechanism_clarity": 5,
"feasibility": 5,
"expected_payoff": 4,
"novelty": 3,
"test_efficiency": 5,
}
assert compute_priority_score(scores) == pytest.approx(4.65, abs=0.01)
def test_rank_hypotheses_sorts_by_priority_score_descending(example_output: dict) -> None:
card_a = deepcopy(example_output["hypotheses"][0])
card_a["hypothesis_id"] = "H-A"
card_b = deepcopy(card_a)
card_b["hypothesis_id"] = "H-B"
card_b["score_components"] = {
"evidence_strength": 2,
"mechanism_clarity": 2,
"feasibility": 2,
"expected_payoff": 2,
"novelty": 2,
"test_efficiency": 2,
}
ranked = rank_hypotheses([card_b, card_a])
assert ranked[0]["hypothesis_id"] == "H-A"
assert ranked[1]["hypothesis_id"] == "H-B"
assert ranked[0]["priority_score"] > ranked[1]["priority_score"]
def test_validate_score_components_reports_missing_key() -> None:
scores = {key: 3 for key in REQUIRED_SCORE_KEYS if key != "novelty"}
errors = validate_score_components(scores)
assert errors
assert any("missing" in err for err in errors)
def test_validate_score_components_rejects_zero() -> None:
scores = {key: 3 for key in REQUIRED_SCORE_KEYS}
scores["novelty"] = 0
errors = validate_score_components(scores)
assert errors
assert any("[1, 5]" in err for err in errors)
def test_validate_score_components_rejects_six() -> None:
scores = {key: 3 for key in REQUIRED_SCORE_KEYS}
scores["novelty"] = 6
errors = validate_score_components(scores)
assert errors
assert any("[1, 5]" in err for err in errors)
def test_rank_hypotheses_raises_value_error_for_invalid_scores(base_hypothesis_card: dict) -> None:
bad = deepcopy(base_hypothesis_card)
bad["score_components"]["novelty"] = 9
with pytest.raises(ValueError, match="invalid score_components"):
rank_hypotheses([bad])
"""Tests for Step H strategy export integration."""
from __future__ import annotations
import json
from copy import deepcopy
from pathlib import Path
import yaml
from pipeline.strategy_exporter import (
build_metadata_json,
build_strategy_yaml,
can_export,
export_candidate,
infer_entry_family,
validate_strategy_yaml,
)
def _hypothesis(base_hypothesis_card: dict, **overrides: object) -> dict:
card = deepcopy(base_hypothesis_card)
card.update(overrides)
return card
class TestInferEntryFamily:
def test_gap_keywords_match(self, base_hypothesis_card: dict) -> None:
card = _hypothesis(
base_hypothesis_card,
thesis="Earnings gap-up with VWAP hold continuation is likely to trend.",
proposed_rule_changes=[
{"component": "entry_logic", "change": "Trade gap-up continuation after VWAP hold."}
],
)
assert infer_entry_family(card) == "gap_up_continuation"
def test_breakout_keywords_match(self, base_hypothesis_card: dict) -> None:
card = _hypothesis(
base_hypothesis_card,
thesis="VCP breakout after contraction increases odds of follow-through.",
proposed_rule_changes=[
{
"component": "entry_logic",
"change": "Require breakout above pivot after VCP pattern.",
}
],
)
assert infer_entry_family(card) == "pivot_breakout"
def test_no_match_returns_none(self, base_hypothesis_card: dict) -> None:
card = _hypothesis(
base_hypothesis_card,
title="Intraday mean reversion filter",
thesis="Mean reversion to prior close after panic selloff.",
proposed_rule_changes=[
{"component": "entry_logic", "change": "Enter after two-sigma deviation fade."}
],
)
assert infer_entry_family(card) is None
class TestCanExport:
def test_pursue_with_valid_family(self, base_hypothesis_card: dict) -> None:
assert can_export(base_hypothesis_card)
def test_park_returns_false(self, base_hypothesis_card: dict) -> None:
card = _hypothesis(base_hypothesis_card, recommendation="park")
assert not can_export(card)
def test_pursue_no_family_returns_false(self, base_hypothesis_card: dict) -> None:
card = _hypothesis(
base_hypothesis_card,
title="Intraday mean reversion filter",
thesis="Mean reversion in oversold names.",
proposed_rule_changes=[
{
"component": "entry_logic",
"change": "Fade extreme dislocations after opening spike.",
}
],
)
assert not can_export(card)
class TestBuildStrategyYaml:
def test_has_all_required_top_level_keys(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
required = {
"id",
"name",
"universe",
"signals",
"risk",
"cost_model",
"validation",
"promotion_gates",
}
assert required.issubset(spec.keys())
def test_signals_exit_has_stop_loss(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
assert isinstance(spec["signals"]["exit"]["stop_loss"], str)
assert spec["signals"]["exit"]["stop_loss"].strip()
def test_signals_exit_has_trailing_or_take_profit(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
exit_rules = spec["signals"]["exit"]
assert bool(exit_rules.get("trailing_stop") or exit_rules.get("take_profit"))
def test_stop_loss_pct_in_range(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
assert 0 < spec["signals"]["exit"]["stop_loss_pct"] <= 0.30
def test_risk_per_trade_in_range(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
assert 0 < spec["risk"]["risk_per_trade"] <= 0.10
def test_validation_method_full_sample(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
assert spec["validation"]["method"] == "full_sample"
assert spec["validation"].get("oos_ratio") is None
def test_gap_up_includes_detection_block(self, base_hypothesis_card: dict) -> None:
card = _hypothesis(
base_hypothesis_card,
thesis="Gap-up continuation after earnings with VWAP hold",
proposed_rule_changes=[
{"component": "entry_logic", "change": "Require gap-up and VWAP hold."}
],
)
spec = build_strategy_yaml(card, "edge_h001")
assert "gap_up_detection" in spec
def test_pivot_breakout_includes_detection_block(self, base_hypothesis_card: dict) -> None:
card = _hypothesis(
base_hypothesis_card,
thesis="VCP breakout with volume expansion",
proposed_rule_changes=[
{"component": "entry_logic", "change": "Require VCP + breakout above pivot."}
],
)
spec = build_strategy_yaml(card, "edge_h001")
assert "vcp_detection" in spec
def test_entry_conditions_non_empty(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
assert spec["signals"]["entry"]["conditions"]
class TestValidateStrategyYaml:
def test_valid_spec_no_errors(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
assert validate_strategy_yaml(spec, "edge_h001") == []
def test_missing_detection_block_error(self, base_hypothesis_card: dict) -> None:
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
spec.pop("gap_up_detection", None)
errors = validate_strategy_yaml(spec, "edge_h001")
assert errors
assert any("gap_up_detection" in err for err in errors)
def test_candidate_contract_import_failure(
self, base_hypothesis_card: dict, monkeypatch
) -> None:
from pipeline import strategy_exporter
spec = build_strategy_yaml(base_hypothesis_card, "edge_h001")
monkeypatch.setattr(
strategy_exporter,
"CANDIDATE_CONTRACT_PATH",
Path("/tmp/does-not-exist/candidate_contract.py"),
)
try:
validate_strategy_yaml(spec, "edge_h001")
except RuntimeError as exc:
assert "candidate_contract.py" in str(exc)
else:
raise AssertionError("RuntimeError was not raised")
class TestExportCandidate:
def test_dry_run_no_files(self, base_hypothesis_card: dict, tmp_path: Path) -> None:
out = export_candidate(base_hypothesis_card, tmp_path, dry_run=True)
assert out is None
assert list(tmp_path.glob("*")) == []
def test_writes_strategy_and_metadata(self, base_hypothesis_card: dict, tmp_path: Path) -> None:
out = export_candidate(base_hypothesis_card, tmp_path, dry_run=False)
assert out is not None
strategy_path = out / "strategy.yaml"
metadata_path = out / "metadata.json"
assert strategy_path.exists()
assert metadata_path.exists()
spec = yaml.safe_load(strategy_path.read_text())
metadata = json.loads(metadata_path.read_text())
assert spec["id"] == metadata["candidate_id"]
assert metadata["interface_version"] == "edge-finder-candidate/v1"
def test_build_metadata_json_contains_provenance(self, base_hypothesis_card: dict) -> None:
metadata = build_metadata_json(base_hypothesis_card, "edge_h001")
assert metadata["interface_version"] == "edge-finder-candidate/v1"
assert metadata["candidate_id"] == "edge_h001"
assert "research_context" in metadata
Related skills
Grill MeHave an agent relentlessly pressure-test their plan or design until every decision branch is resolved and shared understanding is reached.691k196kGrill With DocsPressure-test architecture decisions against their live codebase and living documentation before writing production code.585k196k
Lark CalendarLet their coding agents read, create, and manage events inside Lark (Feishu) Calendar directly from natural language instructions.493kLark ImLet their coding agent send instant notifications, post updates, and receive commands directly inside Lark (Feishu) workspaces.493kLark ContactLet their AI coding agents read and write contacts, send messages, and manage user data inside Lark (Feishu).493kLark AttendanceAutomatically fetch and report daily attendance data from Lark (Feishu) into their own tools or agents.493kHow it compares
Use trade-hypothesis-ideator to scope falsifiable edges from journals; use live market scanner skills when fetching quotes rather than designing experiments.
FAQ
What outputs does trade-hypothesis-ideator produce?
trade-hypothesis-ideator produces one to five ranked hypothesis cards with experiment designs and kill criteria. Pursue-rated cards can export to strategy.yaml and metadata.json using the edge-finder-candidate/v1 format.
What inputs does trade-hypothesis-ideator require?
trade-hypothesis-ideator expects a normalized bundle with market data, trade logs, and journal snippets. The skill generates, critiques, and ranks falsifiable strategy hypotheses from that structured input.
What is edge-finder-candidate/v1 in trade-hypothesis-ideator?
edge-finder-candidate/v1 is the strategy.yaml export schema trade-hypothesis-ideator uses when persisting pursue-worthy hypothesis cards alongside metadata.json for downstream backtest or review tooling.