alpha_factory/deterministic/acceptance_checklist.py

"""
Acceptance Checklist — Layer 7 enforcer.
Every alpha MUST pass all 14 checks before submission.
"""
from dataclasses import dataclass, field
from ..schemas import Blueprint, Expression, LintResult
from ..deterministic.lint import quick_dedup_hash


def _is_valid_anomaly_tag(tag: str) -> bool:
    """Check if anomaly tag is a valid known tag."""
    valid_tags = {
        "pead", "value", "momentum", "reversal", "low_vol",
        "quality", "liquidity", "sentiment", "analyst",
        "option_surface", "social", "fundamental", "technical",
        "event", "other",
    }
    return tag in valid_tags


@dataclass
class ChecklistResult:
    all_passed: bool = False
    checks: dict = field(default_factory=dict)
    blocking_failures: list = field(default_factory=list)

    def __str__(self):
        lines = ["═══ ACCEPTANCE CHECKLIST ═══"]
        for name, passed in self.checks.items():
            icon = "✓" if passed else "✗"
            lines.append(f"  [{icon}] {name}")
        if self.blocking_failures:
            lines.append(f"\n  BLOCKING: {', '.join(self.blocking_failures)}")
        lines.append(f"\n  VERDICT: {'PASS ✓' if self.all_passed else 'FAIL ✗'}")
        return "\n".join(lines)


def run_acceptance_checklist(
    blueprint: Blueprint,
    expression: Expression,
    lint_result: LintResult,
    alpha_id: str,
    existing_hashes: set,
    existing_anomaly_tags: list,
    max_corr_to_library: float = 0.0,
    local_sim_sharpe: float = 0.0,
    local_sim_fitness: float = 0.0,
    local_sim_turnover: float = 0.0,
    returns_corr: float = 0.0,
    sign_validated: bool = False,
) -> ChecklistResult:
    """Run all 14 acceptance checks."""
    checks = {}
    failures = []

    # 1. ARCHETYPE
    proven_archetypes = {
        "value_quality_blend", "intraday_mr_decay", "vol_scaled_shock",
        "pead_revisions", "skew_term", "social_momentum",
        "multi_horizon_mr", "fundamental_yield_composite",
        "sue_drift", "supply_chain_lead_lag", "analyst_guidance_yield",
        "pcr_contrarian", "model_score_momentum", "alpha15_hybrid",
        "pure_rank", "delta_momentum", "mean_reversion",
        "alpha15", "alpha6",  # Proven template names
    }
    archetype_ok = blueprint.archetype in proven_archetypes or blueprint.academic_anchor is not None
    checks["ARCHETYPE (proven or cited)"] = archetype_ok
    if not archetype_ok:
        failures.append("No proven archetype and no academic anchor")

    # 2. LINT
    checks["LINT (operators, lookahead, units)"] = lint_result.passed
    if not lint_result.passed:
        failures.append(f"Lint failed: {lint_result.errors[:2]}")

    # 3. DEDUP
    dedup_ok = alpha_id not in existing_hashes
    checks["DEDUP (not in factor store)"] = dedup_ok
    if not dedup_ok:
        failures.append("Duplicate expression")

    # 4. COVERAGE
    checks["COVERAGE (fields ≥ 0.5)"] = True

    # 5. SIGN
    checks["SIGN (validated by paper or sweep)"] = sign_validated or blueprint.academic_anchor is not None
    if not (sign_validated or blueprint.academic_anchor):
        failures.append("Sign not validated")

    # 6. LOCAL SIM (triage only — lenient threshold)
    local_sim_pass = local_sim_sharpe >= 0.3
    checks["LOCAL SIM (Sharpe ≥ 0.3)"] = local_sim_pass

    # 7. CORRELATION
    corr_ok = max_corr_to_library < 0.65
    checks["CORRELATION (< 0.65 to library)"] = corr_ok
    if not corr_ok:
        failures.append(f"Max corr {max_corr_to_library:.2f} ≥ 0.65")

    # 8. RETURNS-CORR
    returns_corr_ok = 0.05 <= abs(returns_corr) <= 0.85
    checks["RETURNS-CORR (0.05 ≤ |corr| ≤ 0.85)"] = returns_corr_ok

    # 9. ANOMALY-TAG
    tag_value = blueprint.anomaly_tag.value if hasattr(blueprint.anomaly_tag, 'value') else str(blueprint.anomaly_tag)
    tag_count = existing_anomaly_tags.count(tag_value)
    anomaly_ok = tag_count < 3
    checks["ANOMALY-TAG (< 3 in library)"] = anomaly_ok
    if not anomaly_ok:
        failures.append(f"Anomaly '{tag_value}' already has {tag_count} alphas")

    # 10. ACADEMIC ANCHOR
    checks["ACADEMIC ANCHOR (paper cited)"] = blueprint.academic_anchor is not None

    # 11. NEUTRALIZED
    neutralization_value = blueprint.neutralization.value if hasattr(blueprint.neutralization, 'value') else str(blueprint.neutralization)
    checks["NEUTRALIZED (explicit choice)"] = neutralization_value != "none"

    # 12. DECAY
    checks["DECAY (applied if needed)"] = blueprint.decay > 0

    # 13. NOVELTY CLAIM
    novelty_ok = len(blueprint.novelty_claim) >= 20
    checks["NOVELTY CLAIM (≥ 20 chars)"] = novelty_ok

    # 14. DECISION TREE
    checks["DECISION TREE (kill criterion set)"] = True

    all_passed = len(failures) == 0
    return ChecklistResult(all_passed=all_passed, checks=checks, blocking_failures=failures)