Edge Candidate Agent

interface:
  display_name: "Edge Candidate Agent"
  short_description: "Generate edge candidates and Phase I pipeline strategy specs"
  default_prompt: "Generate reproducible edge research tickets and export Phase I-compatible strategy.yaml candidates for trade-strategy-pipeline."

#!/usr/bin/env python3
"""Contract helpers for edge candidate strategy artifacts."""

from __future__ import annotations

from pathlib import Path
from typing import Any

import yaml

INTERFACE_VERSION = "edge-finder-candidate/v1"

SUPPORTED_ENTRY_FAMILIES = {
    "pivot_breakout",
    "gap_up_continuation",
    "panic_reversal",
    "news_reaction",
}

REQUIRED_TOP_LEVEL_KEYS = {
    "id",
    "name",
    "universe",
    "signals",
    "risk",
    "cost_model",
    "validation",
    "promotion_gates",
}


def read_yaml(path: Path) -> dict[str, Any]:
    """Read a YAML file and return a dict payload."""
    payload = yaml.safe_load(path.read_text())
    if not isinstance(payload, dict):
        raise ValueError(f"YAML root must be a mapping: {path}")
    return payload


def write_yaml(path: Path, payload: dict[str, Any]) -> None:
    """Write a dict payload as stable YAML."""
    path.write_text(yaml.safe_dump(payload, sort_keys=False, allow_unicode=False))


def validate_ticket_payload(ticket: dict[str, Any]) -> list[str]:
    """Validate research ticket minimum schema for export."""
    errors: list[str] = []

    for key in ("id", "hypothesis_type", "entry_family"):
        value = ticket.get(key)
        if not isinstance(value, str) or not value.strip():
            errors.append(f"ticket.{key} must be a non-empty string")

    entry_family = ticket.get("entry_family")
    if isinstance(entry_family, str) and entry_family not in SUPPORTED_ENTRY_FAMILIES:
        allowed = ", ".join(sorted(SUPPORTED_ENTRY_FAMILIES))
        errors.append(f"ticket.entry_family must be one of: {allowed}")

    validation = ticket.get("validation")
    if validation is not None:
        if not isinstance(validation, dict):
            errors.append("ticket.validation must be a mapping when provided")
        else:
            method = validation.get("method", "full_sample")
            if method != "full_sample":
                errors.append("ticket.validation.method must be 'full_sample' for Phase I")
            if validation.get("oos_ratio") is not None:
                errors.append("ticket.validation.oos_ratio must be omitted or null for Phase I")

    return errors


def validate_interface_contract(
    spec: dict[str, Any],
    candidate_id: str | None = None,
    stage: str = "phase1",
) -> list[str]:
    """Validate edge-finder-candidate/v1 constraints."""
    errors: list[str] = []

    missing = sorted(REQUIRED_TOP_LEVEL_KEYS - spec.keys())
    if missing:
        errors.append(f"missing required top-level keys: {', '.join(missing)}")

    if candidate_id is not None and spec.get("id") != candidate_id:
        errors.append(
            "candidate directory name must match strategy id "
            f"(candidate_id={candidate_id}, id={spec.get('id')})"
        )

    signals = spec.get("signals")
    if not isinstance(signals, dict):
        errors.append("signals must be a mapping")
        signals = {}

    entry = signals.get("entry")
    if not isinstance(entry, dict):
        errors.append("signals.entry must be a mapping")
        entry = {}

    exit_rules = signals.get("exit")
    if not isinstance(exit_rules, dict):
        errors.append("signals.exit must be a mapping")
        exit_rules = {}

    entry_type = entry.get("type")
    if not isinstance(entry_type, str) or not entry_type.strip():
        errors.append("signals.entry.type must be a non-empty string")
    elif entry_type not in SUPPORTED_ENTRY_FAMILIES:
        allowed = ", ".join(sorted(SUPPORTED_ENTRY_FAMILIES))
        errors.append(f"signals.entry.type must be one of: {allowed}")

    conditions = entry.get("conditions")
    if not isinstance(conditions, list) or len(conditions) == 0:
        errors.append("signals.entry.conditions must be a non-empty list")

    stop_loss = exit_rules.get("stop_loss")
    if not isinstance(stop_loss, str) or not stop_loss.strip():
        errors.append("signals.exit.stop_loss must be a non-empty string")

    trailing_stop = exit_rules.get("trailing_stop")
    take_profit = exit_rules.get("take_profit")
    trailing_ok = isinstance(trailing_stop, str) and bool(trailing_stop.strip())
    take_profit_ok = isinstance(take_profit, str) and bool(take_profit.strip())
    if not (trailing_ok or take_profit_ok):
        errors.append("signals.exit must include trailing_stop or take_profit")

    risk = spec.get("risk")
    if not isinstance(risk, dict):
        errors.append("risk must be a mapping")
        risk = {}

    risk_per_trade = risk.get("risk_per_trade")
    if not _is_number(risk_per_trade) or not (0 < float(risk_per_trade) <= 0.10):
        errors.append("risk.risk_per_trade must satisfy 0 < value <= 0.10")

    max_positions = risk.get("max_positions")
    if isinstance(max_positions, bool) or not isinstance(max_positions, int) or max_positions < 1:
        errors.append("risk.max_positions must be an integer >= 1")

    max_sector_exposure = risk.get("max_sector_exposure")
    if not _is_number(max_sector_exposure) or not (0 < float(max_sector_exposure) <= 1.0):
        errors.append("risk.max_sector_exposure must satisfy 0 < value <= 1.0")

    validation = spec.get("validation")
    if not isinstance(validation, dict):
        errors.append("validation must be a mapping")
        validation = {}

    if stage == "phase1":
        if validation.get("method") != "full_sample":
            errors.append("validation.method must be 'full_sample' in Phase I")
        if validation.get("oos_ratio") is not None:
            errors.append("validation.oos_ratio must be omitted or null in Phase I")

    vcp_detection = spec.get("vcp_detection")
    if vcp_detection is not None and not isinstance(vcp_detection, dict):
        errors.append("vcp_detection must be a mapping when provided")

    gap_detection = spec.get("gap_up_detection")
    if gap_detection is not None and not isinstance(gap_detection, dict):
        errors.append("gap_up_detection must be a mapping when provided")

    if entry_type == "pivot_breakout" and not isinstance(vcp_detection, dict):
        errors.append("pivot_breakout requires vcp_detection block")
    if entry_type == "gap_up_continuation" and not isinstance(gap_detection, dict):
        errors.append("gap_up_continuation requires gap_up_detection block")

    return errors


def _is_number(value: Any) -> bool:
    if isinstance(value, bool):
        return False
    return isinstance(value, (int, float))

#!/usr/bin/env python3
"""Export research tickets to trade-strategy-pipeline candidate artifacts."""

from __future__ import annotations

import argparse
import json
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from candidate_contract import (
    INTERFACE_VERSION,
    read_yaml,
    validate_interface_contract,
    validate_ticket_payload,
    write_yaml,
)

DEFAULT_UNIVERSE = {
    "type": "us_equities",
    "index": "sp500",
    "filters": ["avg_volume > 500_000", "price > 10"],
}

DEFAULT_DATA = {
    "timeframe": "daily",
    "lookback_years": 8,
}

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_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,
    "trailing_start_bars": 0,
}

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",
            "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"],
    },
    "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"],
    },
}

DEFAULT_VCP_DETECTION = {
    "min_contractions": 2,
    "contraction_ratio": 0.75,
    "lookback_window": 120,
    "swing_threshold": 0.05,
    "volume_decline": True,
    "breakout_volume_ratio": 1.5,
    "rolling_window_size": 300,
    "rolling_step_size": 15,
    "rolling_cooldown": 30,
    "atr_multiplier": 1.5,
    "atr_period": 14,
    "min_contraction_days": 5,
    "t1_depth_min": 0.10,
    "t1_depth_max": 0.35,
    "right_shoulder_pct": 0.05,
    "min_pattern_days": 15,
    "max_pattern_days": 325,
    "trend_min_criteria": 6,
}

DEFAULT_GAP_DETECTION = {
    "min_gap_pct": 0.06,
    "volume_ratio": 2.0,
    "avg_volume_window": 50,
    "max_entry_days": 5,
    "breakout_volume_ratio": 1.0,
    "max_stop_pct": 0.10,
    "breakout_close_pos_min": 0.5,
}


class ExportError(Exception):
    """Raised when export cannot proceed safely."""


def deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
    """Recursively merge two dictionaries."""
    merged = deepcopy(base)
    for key, value in override.items():
        if isinstance(value, dict) and isinstance(merged.get(key), dict):
            merged[key] = deep_merge(merged[key], value)
        else:
            merged[key] = value
    return merged


def build_strategy_spec(ticket: dict[str, Any], candidate_id: str) -> dict[str, Any]:
    """Build a Phase I-compatible strategy.yaml payload."""
    entry_family = ticket["entry_family"]

    entry_defaults = DEFAULT_ENTRY_BY_FAMILY[entry_family]
    entry_overrides = ticket.get("entry", {})
    exit_overrides = ticket.get("exit", {})

    if "validation" in ticket:
        validation = ticket["validation"]
        if validation.get("method", "full_sample") != "full_sample":
            raise ExportError("ticket.validation.method must be 'full_sample' for Phase I export")
        if validation.get("oos_ratio") is not None:
            raise ExportError(
                "ticket.validation.oos_ratio must be omitted or null for Phase I export"
            )

    spec: dict[str, Any] = {
        "id": candidate_id,
        "name": ticket.get("name", candidate_id.replace("_", " ").replace("-", " ").title()),
        "description": ticket.get(
            "description",
            f"Auto-generated from ticket {ticket['id']} ({ticket['hypothesis_type']})",
        ),
        "universe": deep_merge(DEFAULT_UNIVERSE, ticket.get("universe", {})),
        "data": deep_merge(DEFAULT_DATA, ticket.get("data", {})),
        "signals": {
            "entry": {
                "type": entry_family,
                "conditions": entry_overrides.get("conditions", entry_defaults["conditions"]),
                "trend_filter": entry_overrides.get("trend_filter", entry_defaults["trend_filter"]),
            },
            "exit": deep_merge(DEFAULT_EXIT, exit_overrides),
        },
        "risk": deep_merge(DEFAULT_RISK, ticket.get("risk", {})),
        "cost_model": deep_merge(DEFAULT_COST_MODEL, ticket.get("cost_model", {})),
        "validation": {"method": "full_sample"},
        "promotion_gates": deep_merge(DEFAULT_PROMOTION_GATES, ticket.get("promotion_gates", {})),
    }

    detection = ticket.get("detection", {})
    if entry_family == "pivot_breakout":
        spec["vcp_detection"] = deep_merge(
            DEFAULT_VCP_DETECTION,
            detection.get("vcp_detection", ticket.get("vcp_detection", {})),
        )
    if entry_family == "gap_up_continuation":
        spec["gap_up_detection"] = deep_merge(
            DEFAULT_GAP_DETECTION,
            detection.get("gap_up_detection", ticket.get("gap_up_detection", {})),
        )

    strategy_overrides = ticket.get("strategy_overrides", {})
    if strategy_overrides:
        if not isinstance(strategy_overrides, dict):
            raise ExportError("ticket.strategy_overrides must be a mapping when provided")
        spec = deep_merge(spec, strategy_overrides)

    return spec


def build_metadata(
    ticket: dict[str, Any],
    candidate_id: str,
    ticket_path: Path,
    generator_version: str,
) -> dict[str, Any]:
    """Build metadata.json payload."""
    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": "edge-candidate-agent",
            "version": generator_version,
            "source_ticket_path": str(ticket_path),
        },
        "research_context": {
            "ticket_id": ticket["id"],
            "hypothesis_type": ticket["hypothesis_type"],
            "mechanism_tag": ticket.get("mechanism_tag", "uncertain"),
            "holding_horizon": ticket.get("holding_horizon", "20D"),
            "entry_family": ticket["entry_family"],
            "regime": ticket.get("regime", "Neutral"),
        },
    }


def export_candidate(
    ticket_path: Path,
    strategies_dir: Path,
    candidate_id: str | None = None,
    force: bool = False,
    dry_run: bool = False,
    generator_version: str = "0.1.0",
) -> tuple[dict[str, Any], dict[str, Any], Path]:
    """Export candidate artifacts and return in-memory payloads."""
    ticket = read_yaml(ticket_path)
    ticket_errors = validate_ticket_payload(ticket)
    if ticket_errors:
        raise ExportError("Ticket validation failed:\n- " + "\n- ".join(ticket_errors))

    resolved_candidate_id = (candidate_id or ticket["id"]).strip()
    if not resolved_candidate_id:
        raise ExportError("candidate_id must not be empty")

    spec = build_strategy_spec(ticket, resolved_candidate_id)
    contract_errors = validate_interface_contract(
        spec,
        candidate_id=resolved_candidate_id,
        stage="phase1",
    )
    if contract_errors:
        raise ExportError(
            "Generated strategy does not satisfy contract:\n- " + "\n- ".join(contract_errors)
        )

    metadata = build_metadata(
        ticket=ticket,
        candidate_id=resolved_candidate_id,
        ticket_path=ticket_path,
        generator_version=generator_version,
    )

    candidate_dir = strategies_dir / resolved_candidate_id
    strategy_path = candidate_dir / "strategy.yaml"
    metadata_path = candidate_dir / "metadata.json"

    if (strategy_path.exists() or metadata_path.exists()) and not force:
        raise ExportError(
            f"candidate artifacts already exist: {candidate_dir} (use --force to overwrite)"
        )

    if not dry_run:
        candidate_dir.mkdir(parents=True, exist_ok=True)
        write_yaml(strategy_path, spec)
        metadata_path.write_text(json.dumps(metadata, indent=2, ensure_ascii=True) + "\n")

    return spec, metadata, candidate_dir


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Export edge research ticket to trade-strategy-pipeline candidate artifacts.",
    )
    parser.add_argument("--ticket", required=True, help="Path to ticket YAML file")
    parser.add_argument(
        "--strategies-dir",
        default="strategies",
        help="Output strategies root directory (default: strategies)",
    )
    parser.add_argument(
        "--candidate-id",
        default=None,
        help="Override candidate id (defaults to ticket.id)",
    )
    parser.add_argument("--force", action="store_true", help="Overwrite existing artifacts")
    parser.add_argument(
        "--dry-run", action="store_true", help="Build and validate without writing files"
    )
    parser.add_argument(
        "--generator-version",
        default="0.1.0",
        help="Version string to write in metadata.json",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    ticket_path = Path(args.ticket).resolve()
    strategies_dir = Path(args.strategies_dir).resolve()

    try:
        spec, metadata, candidate_dir = export_candidate(
            ticket_path=ticket_path,
            strategies_dir=strategies_dir,
            candidate_id=args.candidate_id,
            force=args.force,
            dry_run=args.dry_run,
            generator_version=args.generator_version,
        )
    except (ExportError, ValueError) as exc:
        print(f"[ERROR] {exc}")
        return 1

    print(f"[OK] Candidate id: {spec['id']}")
    print(f"[OK] Entry family: {spec['signals']['entry']['type']}")
    print(f"[OK] Interface version: {metadata['interface_version']}")
    if args.dry_run:
        print("[OK] Dry-run completed. No files were written.")
    else:
        print(f"[OK] Wrote strategy.yaml: {candidate_dir / 'strategy.yaml'}")
        print(f"[OK] Wrote metadata.json: {candidate_dir / 'metadata.json'}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Test configuration for edge-candidate-agent scripts."""

import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

"""Unit tests for auto_detect_candidates.py (pandas-independent parts)."""

from datetime import date
from subprocess import CompletedProcess

import auto_detect_candidates as adc
import pytest


def test_infer_entry_family_from_text_detects_breakout() -> None:
    text = "AI leaders are breaking out to new highs with base contraction"
    assert adc.infer_entry_family_from_text(text) == "pivot_breakout"


def test_infer_entry_family_from_text_detects_gap() -> None:
    text = "Post-earnings gap and follow-through looks strong"
    assert adc.infer_entry_family_from_text(text) == "gap_up_continuation"


def test_normalize_hints_and_hint_boost() -> None:
    hints = adc.normalize_hints(
        [
            {
                "title": "Earnings continuation",
                "preferred_entry_family": "gap_up_continuation",
                "symbols": ["NVDA", "msft"],
            }
        ]
    )
    boost, titles = adc.hint_match_boost("NVDA", "gap_up_continuation", hints)
    assert boost > 0
    assert titles == ["Earnings continuation"]


def test_score_functions_clamp_to_valid_range() -> None:
    breakout = adc.score_breakout_candidate(
        {
            "rs_rank_pct": 1.5,
            "rel_volume": 8.0,
            "close_pos": 1.2,
            "atr_pct": 0.03,
            "close": 120.0,
            "high20_prev": 100.0,
        },
        regime_label="RiskOn",
    )
    gap = adc.score_gap_candidate(
        {
            "gap": 0.2,
            "rel_volume": 8.0,
            "close_pos": 1.2,
            "close": 100.0,
            "ma50": 90.0,
            "ma200": 80.0,
            "atr_pct": 0.04,
        },
        regime_label="RiskOn",
    )
    assert 0.0 <= breakout <= 100.0
    assert 0.0 <= gap <= 100.0


def test_build_ticket_payload_exportable_contains_validation() -> None:
    payload = adc.build_ticket_payload(
        candidate={
            "symbol": "NVDA",
            "entry_family": "pivot_breakout",
            "hypothesis_type": "breakout",
            "priority_score": 88.5,
            "close": 120.0,
            "rel_volume": 2.2,
            "gap": 0.01,
            "close_pos": 0.75,
            "rs_rank_pct": 0.92,
            "conditions": ["close > high20_prev"],
            "trend_filter": ["price > sma_200"],
        },
        as_of_date=date(2026, 2, 20),
        regime_label="RiskOn",
        rank=1,
        market_summary={"pct_above_ma50": 0.65, "avg_pair_corr_20": 0.28, "vol_trend": 0.9},
    )
    assert payload["entry_family"] == "pivot_breakout"
    assert payload["validation"]["method"] == "full_sample"
    assert "entry" in payload


def test_build_ticket_payload_research_only_sets_flag() -> None:
    payload = adc.build_ticket_payload(
        candidate={
            "symbol": "MARKET_BASKET",
            "entry_family": None,
            "hypothesis_type": "regime_shift",
            "priority_score": 72.0,
            "description": "Regime inflection candidate",
        },
        as_of_date=date(2026, 2, 20),
        regime_label="Neutral",
        rank=1,
        market_summary={"pct_above_ma50": 0.48, "avg_pair_corr_20": 0.41, "vol_trend": 1.02},
    )
    assert payload["research_only"] is True
    assert "validation" not in payload
    assert payload["hypothesis_type"] == "regime_shift"


def test_generate_llm_hints_parses_yaml(monkeypatch) -> None:
    stdout = """
hints:
  - title: Momentum leaders
    preferred_entry_family: pivot_breakout
    symbols: [NVDA]
"""

    def fake_run(*args, **kwargs):
        return CompletedProcess(args=args[0], returncode=0, stdout=stdout, stderr="")

    monkeypatch.setattr(adc.subprocess, "run", fake_run)
    hints = adc.generate_llm_hints(
        llm_command="fake-llm-cli",
        as_of_date=date(2026, 2, 20),
        market_summary={"risk_on_score": 70},
        anomalies=[],
    )
    assert len(hints) == 1
    assert hints[0]["entry_family"] == "pivot_breakout"


def test_infer_entry_family_from_text_detects_panic_reversal() -> None:
    text = "Capitulation selloff with oversold reversal signal"
    assert adc.infer_entry_family_from_text(text) == "panic_reversal"


def test_infer_entry_family_from_text_detects_news_reaction() -> None:
    text = "Binary event catalyst with news reaction signal"
    assert adc.infer_entry_family_from_text(text) == "news_reaction"


def test_build_ticket_payload_news_reaction_observation() -> None:
    payload = adc.build_ticket_payload(
        candidate={
            "symbol": "AAPL",
            "entry_family": "news_reaction",
            "hypothesis_type": "news_reaction",
            "priority_score": 75.0,
            "close": 180.0,
            "rel_volume": 3.5,
            "gap": 0.08,
            "close_pos": 0.65,
            "rs_rank_pct": 0.50,
            "reaction_1d": -0.09,
            "conditions": ["abs_reaction_1d >= 0.06"],
            "trend_filter": ["validate_follow_through_d2"],
        },
        as_of_date=date(2026, 2, 20),
        regime_label="Neutral",
        rank=1,
        market_summary={"pct_above_ma50": 0.50},
    )
    assert payload["entry_family"] == "news_reaction"
    assert payload["observation"]["abs_reaction_1d"] == 0.09
    assert payload["observation"]["reaction_direction"] == "down"


def test_build_ticket_payload_panic_reversal_hypothesis() -> None:
    payload = adc.build_ticket_payload(
        candidate={
            "symbol": "TSLA",
            "entry_family": "panic_reversal",
            "hypothesis_type": "panic_reversal",
            "priority_score": 80.0,
            "close": 200.0,
            "rel_volume": 2.5,
            "gap": -0.08,
            "close_pos": 0.30,
            "rs_rank_pct": 0.20,
            "conditions": ["ret_1d <= -0.07"],
            "trend_filter": ["price > sma_200 * 0.85"],
        },
        as_of_date=date(2026, 2, 20),
        regime_label="RiskOff",
        rank=1,
        market_summary={"pct_above_ma50": 0.30},
    )
    assert payload["entry_family"] == "panic_reversal"
    assert "rebound probability" in payload["hypothesis"]


def test_scan_news_reaction_candidates_canonical_conditions() -> None:
    """Verify scanner emits canonical threshold conditions, not raw values."""
    pytest.importorskip("pandas")
    import pandas as pd

    news = pd.DataFrame(
        {
            "symbol": ["AAPL", "MSFT"],
            "timestamp": pd.to_datetime(["2026-02-20", "2026-02-20"], utc=True),
            "reaction_1d": [0.12, -0.08],
        }
    )
    candidates, available = adc.scan_news_reaction_candidates(
        news_table=news,
        target_date=date(2026, 2, 20),
        tradable=None,
        top_n=4,
    )
    assert available is True
    assert len(candidates) == 2
    # All candidates must use canonical thresholds, not observed values
    for c in candidates:
        assert c["entry_family"] == "news_reaction"
        assert "abs_reaction_1d >= 0.06" in c["conditions"]
        assert "rel_volume >= 2.0" in c["conditions"]
        # Must NOT contain raw observed value like "reaction_1d=0.12"
        for cond in c["conditions"]:
            assert not cond.startswith("reaction_1d=")


def test_scan_news_reaction_candidates_with_tradable_join() -> None:
    """Verify tradable join enriches candidates with OHLCV columns."""
    pytest.importorskip("pandas")
    import pandas as pd

    news = pd.DataFrame(
        {
            "symbol": ["AAPL"],
            "timestamp": pd.to_datetime(["2026-02-20"], utc=True),
            "reaction_1d": [0.10],
        }
    )
    tradable = pd.DataFrame(
        {
            "symbol": ["AAPL", "MSFT"],
            "rel_volume": [3.0, 1.5],
            "close_pos": [0.8, 0.4],
            "atr_pct": [0.025, 0.020],
        }
    )
    candidates, _ = adc.scan_news_reaction_candidates(
        news_table=news,
        target_date=date(2026, 2, 20),
        tradable=tradable,
        top_n=4,
    )
    assert len(candidates) == 1
    assert candidates[0]["symbol"] == "AAPL"
    assert candidates[0]["entry_family"] == "news_reaction"


def test_generate_llm_hints_raises_on_failure(monkeypatch) -> None:
    def fake_run(*args, **kwargs):
        return CompletedProcess(args=args[0], returncode=1, stdout="", stderr="failure")

    monkeypatch.setattr(adc.subprocess, "run", fake_run)
    with pytest.raises(adc.AutoDetectError, match="LLM ideas command failed"):
        adc.generate_llm_hints(
            llm_command="fake-llm-cli",
            as_of_date=date(2026, 2, 20),
            market_summary={"risk_on_score": 70},
            anomalies=[],
        )

"""Unit tests for edge candidate interface checks."""

from copy import deepcopy

from candidate_contract import validate_interface_contract, validate_ticket_payload


def build_valid_spec() -> dict:
    return {
        "id": "edge_vcp_breakout_v1",
        "name": "Edge VCP Breakout v1",
        "description": "Test strategy",
        "universe": {
            "type": "us_equities",
            "index": "sp500",
            "filters": ["avg_volume > 500_000", "price > 10"],
        },
        "signals": {
            "entry": {
                "type": "pivot_breakout",
                "conditions": ["vcp_pattern_detected"],
                "trend_filter": ["price > sma_200"],
            },
            "exit": {
                "stop_loss": "7% below entry",
                "trailing_stop": "below 21-day EMA",
                "take_profit": "risk_reward_3x",
            },
        },
        "risk": {
            "risk_per_trade": 0.01,
            "max_positions": 5,
            "max_sector_exposure": 0.30,
        },
        "cost_model": {
            "commission_per_share": 0.0,
            "slippage_bps": 5,
        },
        "validation": {"method": "full_sample"},
        "promotion_gates": {
            "min_trades": 200,
            "max_drawdown": 0.15,
            "sharpe": 1.0,
            "profit_factor": 1.2,
        },
        "vcp_detection": {
            "min_contractions": 2,
            "contraction_ratio": 0.75,
        },
    }


def test_valid_phase1_spec_passes() -> None:
    spec = build_valid_spec()
    errors = validate_interface_contract(spec, candidate_id=spec["id"], stage="phase1")
    assert errors == []


def test_phase1_rejects_walk_forward() -> None:
    spec = build_valid_spec()
    spec["validation"] = {"method": "walk_forward", "oos_ratio": 0.3}
    errors = validate_interface_contract(spec, candidate_id=spec["id"], stage="phase1")
    assert any("validation.method" in error for error in errors)
    assert any("validation.oos_ratio" in error for error in errors)


def test_entry_family_requires_matching_detection_block() -> None:
    spec = build_valid_spec()
    spec["signals"]["entry"]["type"] = "gap_up_continuation"
    errors = validate_interface_contract(spec, candidate_id=spec["id"], stage="phase1")
    assert any("gap_up_continuation requires gap_up_detection" in error for error in errors)


def test_candidate_id_mismatch_is_rejected() -> None:
    spec = build_valid_spec()
    errors = validate_interface_contract(spec, candidate_id="different_id", stage="phase1")
    assert any("candidate directory name must match strategy id" in error for error in errors)


def test_risk_bounds_violation_is_rejected() -> None:
    spec = deepcopy(build_valid_spec())
    spec["risk"]["risk_per_trade"] = 0.15
    errors = validate_interface_contract(spec, candidate_id=spec["id"], stage="phase1")
    assert any("risk.risk_per_trade" in error for error in errors)


def test_validate_ticket_payload_accepts_valid_minimum() -> None:
    ticket = {
        "id": "edge_vcp_breakout_v1",
        "hypothesis_type": "breakout",
        "entry_family": "pivot_breakout",
    }
    assert validate_ticket_payload(ticket) == []


def test_validate_ticket_payload_rejects_missing_required_fields() -> None:
    ticket = {"id": "", "hypothesis_type": "breakout"}
    errors = validate_ticket_payload(ticket)
    assert any("ticket.id" in error for error in errors)
    assert any("ticket.entry_family" in error for error in errors)


def test_validate_ticket_payload_rejects_unsupported_entry_family() -> None:
    ticket = {
        "id": "edge_momentum_v1",
        "hypothesis_type": "momentum",
        "entry_family": "momentum_followthrough",
    }
    errors = validate_ticket_payload(ticket)
    assert any("ticket.entry_family must be one of" in error for error in errors)


def test_validate_ticket_payload_rejects_non_phase1_validation_method() -> None:
    ticket = {
        "id": "edge_vcp_breakout_v1",
        "hypothesis_type": "breakout",
        "entry_family": "pivot_breakout",
        "validation": {"method": "walk_forward"},
    }
    errors = validate_ticket_payload(ticket)
    assert any("ticket.validation.method" in error for error in errors)


def test_validate_ticket_payload_rejects_non_null_oos_ratio() -> None:
    ticket = {
        "id": "edge_vcp_breakout_v1",
        "hypothesis_type": "breakout",
        "entry_family": "pivot_breakout",
        "validation": {"method": "full_sample", "oos_ratio": 0.3},
    }
    errors = validate_ticket_payload(ticket)
    assert any("ticket.validation.oos_ratio" in error for error in errors)

"""Unit tests for exporting ticket payloads to candidate artifacts."""

import json
from pathlib import Path

import pytest
import yaml
from candidate_contract import INTERFACE_VERSION, read_yaml, validate_interface_contract
from export_candidate import ExportError, deep_merge, export_candidate


def write_ticket(path: Path, payload: dict) -> None:
    path.write_text(yaml.safe_dump(payload, sort_keys=False, allow_unicode=False))


def test_export_pivot_breakout_candidate(tmp_path: Path) -> None:
    ticket_path = tmp_path / "ticket.yaml"
    write_ticket(
        ticket_path,
        {
            "id": "edge_vcp_breakout_v1",
            "hypothesis_type": "breakout",
            "entry_family": "pivot_breakout",
            "name": "Edge VCP Breakout v1",
            "description": "Pivot breakout test candidate",
        },
    )

    strategies_dir = tmp_path / "strategies"
    spec, metadata, candidate_dir = export_candidate(
        ticket_path=ticket_path,
        strategies_dir=strategies_dir,
    )

    assert candidate_dir == strategies_dir / "edge_vcp_breakout_v1"
    assert (candidate_dir / "strategy.yaml").exists()
    assert (candidate_dir / "metadata.json").exists()

    loaded_spec = read_yaml(candidate_dir / "strategy.yaml")
    errors = validate_interface_contract(
        loaded_spec, candidate_id="edge_vcp_breakout_v1", stage="phase1"
    )
    assert errors == []
    assert loaded_spec["signals"]["entry"]["type"] == "pivot_breakout"
    assert "vcp_detection" in loaded_spec

    loaded_metadata = json.loads((candidate_dir / "metadata.json").read_text())
    assert loaded_metadata["interface_version"] == INTERFACE_VERSION
    assert loaded_metadata["candidate_id"] == "edge_vcp_breakout_v1"
    assert metadata["candidate_id"] == "edge_vcp_breakout_v1"
    assert spec["id"] == "edge_vcp_breakout_v1"


def test_export_gap_candidate_with_override_id(tmp_path: Path) -> None:
    ticket_path = tmp_path / "ticket.yaml"
    write_ticket(
        ticket_path,
        {
            "id": "edge_gap_followthrough_v1",
            "hypothesis_type": "earnings_drift",
            "entry_family": "gap_up_continuation",
            "name": "Edge Gap Followthrough v1",
            "description": "Gap continuation test candidate",
        },
    )

    strategies_dir = tmp_path / "strategies"
    spec, _, candidate_dir = export_candidate(
        ticket_path=ticket_path,
        strategies_dir=strategies_dir,
        candidate_id="edge_gap_override_v1",
    )

    assert candidate_dir == strategies_dir / "edge_gap_override_v1"
    assert spec["id"] == "edge_gap_override_v1"
    loaded_spec = read_yaml(candidate_dir / "strategy.yaml")
    assert loaded_spec["signals"]["entry"]["type"] == "gap_up_continuation"
    assert "gap_up_detection" in loaded_spec


def test_export_rejects_unsupported_entry_family(tmp_path: Path) -> None:
    ticket_path = tmp_path / "ticket.yaml"
    write_ticket(
        ticket_path,
        {
            "id": "edge_momentum_v1",
            "hypothesis_type": "momentum",
            "entry_family": "momentum_followthrough",
        },
    )

    strategies_dir = tmp_path / "strategies"
    with pytest.raises(ExportError, match="ticket.entry_family must be one of"):
        export_candidate(ticket_path=ticket_path, strategies_dir=strategies_dir)


def test_export_dry_run_does_not_write_artifacts(tmp_path: Path) -> None:
    ticket_path = tmp_path / "ticket.yaml"
    write_ticket(
        ticket_path,
        {
            "id": "edge_vcp_dry_run_v1",
            "hypothesis_type": "breakout",
            "entry_family": "pivot_breakout",
        },
    )

    strategies_dir = tmp_path / "strategies"
    spec, metadata, candidate_dir = export_candidate(
        ticket_path=ticket_path,
        strategies_dir=strategies_dir,
        dry_run=True,
    )

    assert spec["id"] == "edge_vcp_dry_run_v1"
    assert metadata["candidate_id"] == "edge_vcp_dry_run_v1"
    assert not candidate_dir.exists()
    assert not (candidate_dir / "strategy.yaml").exists()
    assert not (candidate_dir / "metadata.json").exists()


def test_export_force_overwrites_existing_artifacts(tmp_path: Path) -> None:
    ticket_path = tmp_path / "ticket.yaml"
    write_ticket(
        ticket_path,
        {
            "id": "edge_vcp_force_v1",
            "hypothesis_type": "breakout",
            "entry_family": "pivot_breakout",
            "description": "first description",
        },
    )

    strategies_dir = tmp_path / "strategies"
    export_candidate(ticket_path=ticket_path, strategies_dir=strategies_dir)

    write_ticket(
        ticket_path,
        {
            "id": "edge_vcp_force_v1",
            "hypothesis_type": "breakout",
            "entry_family": "pivot_breakout",
            "description": "updated description",
        },
    )

    with pytest.raises(ExportError, match="already exist"):
        export_candidate(ticket_path=ticket_path, strategies_dir=strategies_dir, force=False)

    export_candidate(ticket_path=ticket_path, strategies_dir=strategies_dir, force=True)
    loaded_spec = read_yaml(strategies_dir / "edge_vcp_force_v1" / "strategy.yaml")
    assert loaded_spec["description"] == "updated description"


def test_deep_merge_handles_nested_and_scalar_overrides() -> None:
    base = {
        "a": {"x": 1, "y": 2},
        "b": 10,
    }
    override = {
        "a": {"y": 20, "z": 30},
        "b": {"nested": True},
        "c": 99,
    }
    merged = deep_merge(base, override)

    assert merged["a"] == {"x": 1, "y": 20, "z": 30}
    assert merged["b"] == {"nested": True}
    assert merged["c"] == 99


def test_deep_merge_with_empty_override_returns_same_structure() -> None:
    base = {"a": {"x": 1}, "b": 2}
    merged = deep_merge(base, {})
    assert merged == base

"""Unit tests for validate_candidate.py."""

import subprocess
import sys
from pathlib import Path

import validate_candidate
import yaml


def build_valid_strategy(strategy_id: str) -> dict:
    return {
        "id": strategy_id,
        "name": "Validation Test Strategy",
        "description": "test",
        "universe": {
            "type": "us_equities",
            "index": "sp500",
            "filters": ["avg_volume > 500_000", "price > 10"],
        },
        "signals": {
            "entry": {
                "type": "pivot_breakout",
                "conditions": ["vcp_pattern_detected"],
                "trend_filter": ["price > sma_200"],
            },
            "exit": {
                "stop_loss": "7% below entry",
                "trailing_stop": "below 21-day EMA",
                "take_profit": "risk_reward_3x",
            },
        },
        "risk": {
            "risk_per_trade": 0.01,
            "max_positions": 5,
            "max_sector_exposure": 0.30,
        },
        "cost_model": {
            "commission_per_share": 0.0,
            "slippage_bps": 5,
        },
        "validation": {"method": "full_sample"},
        "promotion_gates": {
            "min_trades": 200,
            "max_drawdown": 0.15,
            "sharpe": 1.0,
            "profit_factor": 1.2,
        },
        "vcp_detection": {"min_contractions": 2, "contraction_ratio": 0.75},
    }


def write_strategy(path: Path, strategy_id: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = build_valid_strategy(strategy_id)
    path.write_text(yaml.safe_dump(payload, sort_keys=False, allow_unicode=False))


def test_validate_with_pipeline_schema_requires_src_dir(tmp_path: Path) -> None:
    strategy_path = tmp_path / "strategy.yaml"
    write_strategy(strategy_path, "edge_schema_missing_src_v1")
    errors = validate_candidate.validate_with_pipeline_schema(
        strategy_path=strategy_path,
        pipeline_root=tmp_path,
        stage="phase1",
    )
    assert any("pipeline source directory not found" in error for error in errors)


def test_validate_with_pipeline_schema_delegates_to_uv(
    tmp_path: Path,
    monkeypatch,
) -> None:
    strategy_path = tmp_path / "strategy.yaml"
    write_strategy(strategy_path, "edge_schema_delegate_v1")
    (tmp_path / "src").mkdir()

    calls = {}

    def fake_validate_with_pipeline_uv(
        strategy_path: Path, pipeline_root: Path, stage: str
    ) -> list[str]:
        calls["strategy_path"] = strategy_path
        calls["pipeline_root"] = pipeline_root
        calls["stage"] = stage
        return []

    monkeypatch.setattr(
        validate_candidate, "validate_with_pipeline_uv", fake_validate_with_pipeline_uv
    )
    errors = validate_candidate.validate_with_pipeline_schema(
        strategy_path=strategy_path,
        pipeline_root=tmp_path,
        stage="phase1",
    )

    assert errors == []
    assert calls["strategy_path"] == strategy_path
    assert calls["pipeline_root"] == tmp_path
    assert calls["stage"] == "phase1"


def test_validate_with_pipeline_uv_handles_subprocess_failure(
    tmp_path: Path,
    monkeypatch,
) -> None:
    strategy_path = tmp_path / "strategy.yaml"
    write_strategy(strategy_path, "edge_uv_fail_v1")

    def fake_run(*args, **kwargs):
        return subprocess.CompletedProcess(args=args[0], returncode=1, stdout="", stderr="boom")

    monkeypatch.setattr(validate_candidate.subprocess, "run", fake_run)
    errors = validate_candidate.validate_with_pipeline_uv(
        strategy_path=strategy_path,
        pipeline_root=tmp_path,
        stage="phase1",
    )
    assert any("uv pipeline validation failed" in error for error in errors)


def test_validate_with_pipeline_uv_parses_error_list(
    tmp_path: Path,
    monkeypatch,
) -> None:
    strategy_path = tmp_path / "strategy.yaml"
    write_strategy(strategy_path, "edge_uv_json_v1")

    def fake_run(*args, **kwargs):
        return subprocess.CompletedProcess(
            args=args[0],
            returncode=0,
            stdout='{"errors":["phase1 violation"]}\n',
            stderr="",
        )

    monkeypatch.setattr(validate_candidate.subprocess, "run", fake_run)
    errors = validate_candidate.validate_with_pipeline_uv(
        strategy_path=strategy_path,
        pipeline_root=tmp_path,
        stage="phase1",
    )
    assert errors == ["phase1 violation"]


def test_validate_with_pipeline_uv_rejects_invalid_json(
    tmp_path: Path,
    monkeypatch,
) -> None:
    strategy_path = tmp_path / "strategy.yaml"
    write_strategy(strategy_path, "edge_uv_bad_json_v1")

    def fake_run(*args, **kwargs):
        return subprocess.CompletedProcess(
            args=args[0],
            returncode=0,
            stdout="not-json\n",
            stderr="",
        )

    monkeypatch.setattr(validate_candidate.subprocess, "run", fake_run)
    errors = validate_candidate.validate_with_pipeline_uv(
        strategy_path=strategy_path,
        pipeline_root=tmp_path,
        stage="phase1",
    )
    assert any("failed to parse uv validation output as JSON" in error for error in errors)


def test_main_returns_error_for_missing_strategy(monkeypatch, capsys, tmp_path: Path) -> None:
    missing_path = tmp_path / "missing.yaml"
    monkeypatch.setattr(sys, "argv", ["validate_candidate.py", "--strategy", str(missing_path)])
    rc = validate_candidate.main()
    out = capsys.readouterr().out
    assert rc == 1
    assert "strategy not found" in out


def test_main_interface_only_success(monkeypatch, capsys, tmp_path: Path) -> None:
    strategy_id = "edge_main_success_v1"
    strategy_path = tmp_path / strategy_id / "strategy.yaml"
    write_strategy(strategy_path, strategy_id)

    monkeypatch.setattr(sys, "argv", ["validate_candidate.py", "--strategy", str(strategy_path)])
    rc = validate_candidate.main()
    out = capsys.readouterr().out
    assert rc == 0
    assert "Candidate validation passed" in out


def test_main_strict_pipeline_requires_root(monkeypatch, capsys, tmp_path: Path) -> None:
    strategy_id = "edge_main_strict_v1"
    strategy_path = tmp_path / strategy_id / "strategy.yaml"
    write_strategy(strategy_path, strategy_id)

    monkeypatch.setattr(
        sys,
        "argv",
        ["validate_candidate.py", "--strategy", str(strategy_path), "--strict-pipeline"],
    )
    rc = validate_candidate.main()
    out = capsys.readouterr().out
    assert rc == 1
    assert "--strict-pipeline requires --pipeline-root" in out


def test_main_pipeline_root_uses_schema_validator(monkeypatch, capsys, tmp_path: Path) -> None:
    strategy_id = "edge_main_pipeline_v1"
    strategy_path = tmp_path / strategy_id / "strategy.yaml"
    write_strategy(strategy_path, strategy_id)

    def fake_validate_with_pipeline_schema(
        strategy_path: Path, pipeline_root: Path, stage: str
    ) -> list[str]:
        assert stage == "phase1"
        return ["forced pipeline error"]

    monkeypatch.setattr(
        validate_candidate,
        "validate_with_pipeline_schema",
        fake_validate_with_pipeline_schema,
    )
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "validate_candidate.py",
            "--strategy",
            str(strategy_path),
            "--pipeline-root",
            str(tmp_path),
            "--stage",
            "phase1",
        ],
    )
    rc = validate_candidate.main()
    out = capsys.readouterr().out
    assert rc == 1
    assert "forced pipeline error" in out

#!/usr/bin/env python3
"""Validate candidate strategy artifacts against interface and pipeline rules."""

from __future__ import annotations

import argparse
import json
import os
import subprocess
from pathlib import Path

from candidate_contract import read_yaml, validate_interface_contract


def validate_with_pipeline_schema(
    strategy_path: Path,
    pipeline_root: Path,
    stage: str,
) -> list[str]:
    """Validate with trade-strategy-pipeline source tree and stage rules."""
    src_dir = pipeline_root / "src"
    if not src_dir.exists():
        return [f"pipeline source directory not found: {src_dir}"]
    return validate_with_pipeline_uv(
        strategy_path=strategy_path,
        pipeline_root=pipeline_root,
        stage=stage,
    )


def validate_with_pipeline_uv(
    strategy_path: Path,
    pipeline_root: Path,
    stage: str,
) -> list[str]:
    """Validate by executing inside pipeline's `uv run` environment."""
    snippet = "\n".join(
        [
            "import json",
            "from pathlib import Path",
            "import yaml",
            "from pipeline.spec.schema import StrategySpec",
            "from pipeline.spec.validator import validate_spec",
            f"strategy_path = Path({strategy_path.as_posix()!r})",
            f"stage = {stage!r}",
            "with open(strategy_path) as f:",
            "    payload = yaml.safe_load(f)",
            "spec = StrategySpec(**payload)",
            "errors = validate_spec(spec, stage=stage)",
            "print(json.dumps({'errors': errors}))",
        ]
    )

    result = subprocess.run(  # nosec B607 - uv is a known local tool
        ["uv", "run", "python", "-c", snippet],
        cwd=str(pipeline_root),
        env=_uv_env(),
        capture_output=True,
        text=True,
    )

    if result.returncode != 0:
        detail = (result.stderr or result.stdout).strip()
        return [f"uv pipeline validation failed: {detail}"]

    lines = [line for line in result.stdout.splitlines() if line.strip()]
    if not lines:
        return ["uv pipeline validation produced no output"]

    try:
        payload = json.loads(lines[-1])
    except json.JSONDecodeError as exc:
        return [f"failed to parse uv validation output as JSON: {exc}"]

    errors = payload.get("errors", [])
    if not isinstance(errors, list):
        return ["uv validation output missing 'errors' list"]
    return [str(error) for error in errors]


def _uv_env() -> dict[str, str]:
    """Return subprocess environment with writable uv cache."""
    env = os.environ.copy()
    env.setdefault("UV_CACHE_DIR", "/tmp/uv-cache-edge-candidate-agent")  # nosec B108
    return env


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Validate strategy candidate against edge-finder interface and pipeline constraints.",
    )
    parser.add_argument("--strategy", required=True, help="Path to strategy.yaml")
    parser.add_argument(
        "--candidate-id",
        default=None,
        help="Candidate id override (defaults to strategy parent directory name)",
    )
    parser.add_argument(
        "--stage", default="phase1", help="Pipeline validation stage (default: phase1)"
    )
    parser.add_argument(
        "--pipeline-root",
        default=None,
        help="Path to trade-strategy-pipeline repository root for strict schema validation",
    )
    parser.add_argument(
        "--strict-pipeline",
        action="store_true",
        help="Fail when --pipeline-root is not provided",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    strategy_path = Path(args.strategy).resolve()
    if not strategy_path.exists():
        print(f"[ERROR] strategy not found: {strategy_path}")
        return 1

    try:
        strategy_payload = read_yaml(strategy_path)
    except (OSError, ValueError) as exc:
        print(f"[ERROR] failed to read strategy YAML: {exc}")
        return 1

    candidate_id = args.candidate_id or strategy_path.parent.name
    errors = validate_interface_contract(
        strategy_payload,
        candidate_id=candidate_id,
        stage=args.stage,
    )

    pipeline_errors: list[str] = []
    if args.pipeline_root:
        pipeline_root = Path(args.pipeline_root).resolve()
        pipeline_errors = validate_with_pipeline_schema(
            strategy_path=strategy_path,
            pipeline_root=pipeline_root,
            stage=args.stage,
        )
    elif args.strict_pipeline:
        pipeline_errors = ["--strict-pipeline requires --pipeline-root"]

    all_errors = errors + pipeline_errors
    if all_errors:
        print("[ERROR] Candidate validation failed:")
        for error in all_errors:
            print(f"  - {error}")
        return 1

    print(f"[OK] Candidate validation passed: {strategy_path}")
    print(f"[OK] candidate_id={candidate_id}")
    if args.pipeline_root:
        print(f"[OK] pipeline schema validated against stage={args.stage}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())