"""
Smoke tests for analysis/bias.py — AI override + BiasResolver.

Covers:
  - resolve() unambiguous algo path (no AI call)
  - resolve() ambiguous algo path with AI happy/parse-error/timeout/credit
  - cache TTL hit / miss (manipulate computed_at)
  - invalidate(symbol) and invalidate(all)
  - AI validation: garbage bias -> fallback algo, allowed BOTH coerced to NONE
                   when bias=NEUTRO (V14 coherence, V15 line 257-259),
                   h1_compatibility clamped to [0.3, 1.0]
  - persistence roundtrip BiasData -> BiasEntry -> BiasData
  - V15 router parity: df4=None -> NEUTRO BOTH 0.5

FakeAIClient implements ask() (BiasResolver calls .ask, not .ask_for_decision).

Run:
    cd ~/apex_v16
    python -m tests.test_bias_resolver
"""

from __future__ import annotations

import asyncio
import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from analysis.bias import (
    AI_BIAS_PROMPT,
    BiasData,
    BiasResolver,
    _bias_data_to_entry,
    _entry_to_bias_data,
    compute_ai_bias,
    compute_algo_bias,
)
from brain.ai_client import AIResponse
from core.contracts import utc_now
from persistence.state_store import BiasEntry, SessionState


def _ok(label: str) -> None:
    print(f"  ok  {label}")


# ============================================================
# FAKES
# ============================================================

class FakeAIClient:
    """
    Mimics AIClient.ask: returns canned AIResponse, tracks calls.
    """
    def __init__(self, canned: AIResponse | None = None) -> None:
        self.canned = canned or AIResponse(text=None, error_kind="unknown")
        self.calls: list[dict] = []

    async def ask(self, prompt: str, temperature: float = 0.2,
                  max_tokens: int | None = None) -> AIResponse:
        self.calls.append({
            "temperature": temperature,
            "prompt_preview": prompt[:60],
        })
        return self.canned


def make_ai_response(payload: dict) -> AIResponse:
    return AIResponse(text=json.dumps(payload), attempts=1)


# ============================================================
# DF FIXTURES — synthetic H4 series
# ============================================================

def df4_unambiguous_bullish() -> pd.DataFrame:
    """Noisy uptrend: EMA20>EMA50, struct bullish, RSI in (50, 80)."""
    rows, base = [], 100.0
    pattern = [+0.3, +0.2, -0.4, +0.5, +0.3, -0.5, +0.4, +0.3, -0.4, +0.5] * 3
    for delta in pattern:
        o = base
        c = base + delta
        rows.append({
            "open": o, "high": max(o, c) + 0.1,
            "low": min(o, c) - 0.1, "close": c, "volume": 1000,
        })
        base = c
    return pd.DataFrame(rows)


def df4_ambiguous() -> pd.DataFrame:
    """20 bars up then 8 bars down: EMA bullish but recent struct LH/LL."""
    closes = [100.0 + i * 0.5 for i in range(20)] + \
             [110.0 - i * 0.6 for i in range(1, 9)]
    rows = []
    for i in range(len(closes)):
        nxt = closes[min(i + 1, len(closes) - 1)]
        rows.append({
            "open": closes[i], "close": nxt,
            "high": max(closes[i], nxt) + 0.05,
            "low": min(closes[i], nxt) - 0.05,
            "volume": 1000,
        })
    return pd.DataFrame(rows)


def fresh_state() -> SessionState:
    return SessionState()


# ============================================================
# 1. Unambiguous: no AI call
# ============================================================

def test_resolve_unambiguous_skips_ai():
    state = fresh_state()
    ai = FakeAIClient()
    resolver = BiasResolver(ai_client=ai, state=state)
    df4 = df4_unambiguous_bullish()
    bd = asyncio.run(resolver.resolve("MES", df4))
    assert bd.bias == "RIALZISTA", f"got {bd.bias} | {bd.h1_reason}"
    assert bd.allowed_direction == "BUY"
    assert bd.ambiguous is False
    assert len(ai.calls) == 0, "AI must NOT be called on unambiguous algo"
    assert state.brain.bias_calls_count == 0
    assert "MES" in state.bias_cache.entries
    _ok("resolve: unambiguous algo -> no AI call, cache populated")


# ============================================================
# 2. Ambiguous + AI happy path
# ============================================================

def test_resolve_ambiguous_calls_ai_happy():
    state = fresh_state()
    ai = FakeAIClient(canned=make_ai_response({
        "bias": "RIALZISTA",
        "allowed_direction": "BUY",
        "h1_compatibility": 0.85,
        "reason": "AI sees bullish continuation",
    }))
    resolver = BiasResolver(ai_client=ai, state=state)
    bd = asyncio.run(resolver.resolve("MES", df4_ambiguous()))
    assert bd.bias == "RIALZISTA"
    assert bd.allowed_direction == "BUY"
    assert bd.h1_compatibility == 0.85
    assert bd.h1_reason.startswith("AI: "), f"reason={bd.h1_reason}"
    assert len(ai.calls) == 1
    assert ai.calls[0]["temperature"] == 0.7   # V15 calibration preserved
    assert state.brain.bias_calls_count == 1
    _ok("resolve: ambiguous + AI valid -> override applied, counter=1, temp=0.7")


# ============================================================
# 3. Ambiguous + AI parse error -> fallback algo, counter still bumped
# ============================================================

def test_resolve_ambiguous_ai_parse_error_fallback():
    state = fresh_state()
    ai = FakeAIClient(canned=AIResponse(text="this is not JSON at all", attempts=1))
    resolver = BiasResolver(ai_client=ai, state=state)
    bd = asyncio.run(resolver.resolve("MES", df4_ambiguous()))
    # Fallback to algo: ambiguous algo on this fixture is NEUTRO
    assert bd.bias == "NEUTRO"
    assert not bd.h1_reason.startswith("AI: ")
    assert state.brain.bias_calls_count == 1, "counter must bump even on parse error"
    _ok("resolve: AI parse error -> fallback algo, counter still bumped")


# ============================================================
# 4. Ambiguous + AI overload (text=None) -> fallback algo
# ============================================================

def test_resolve_ambiguous_ai_timeout_fallback():
    state = fresh_state()
    ai = FakeAIClient(canned=AIResponse(text=None, error_kind="overload", attempts=3))
    resolver = BiasResolver(ai_client=ai, state=state)
    bd = asyncio.run(resolver.resolve("MES", df4_ambiguous()))
    assert bd.bias == "NEUTRO"
    assert state.brain.bias_calls_count == 1
    _ok("resolve: AI overload (text=None) -> fallback algo")


# ============================================================
# 5. Ambiguous + AI credit -> fallback algo
# ============================================================

def test_resolve_ambiguous_ai_credit_fallback():
    state = fresh_state()
    ai = FakeAIClient(canned=AIResponse(
        text=None, error_kind="credit", attempts=1,
        error_detail="credit_balance_too_low",
    ))
    resolver = BiasResolver(ai_client=ai, state=state)
    bd = asyncio.run(resolver.resolve("MES", df4_ambiguous()))
    assert bd.bias == "NEUTRO"
    assert state.brain.bias_calls_count == 1
    _ok("resolve: AI credit-exhausted -> fallback algo (no raise)")


# ============================================================
# 6. Cache hit within TTL: no AI, no algo recompute
# ============================================================

def test_cache_hit_within_ttl():
    state = fresh_state()
    ai = FakeAIClient()
    resolver = BiasResolver(ai_client=ai, state=state)
    # First resolve populates cache
    asyncio.run(resolver.resolve("MES", df4_unambiguous_bullish()))
    cached_entry = state.bias_cache.entries["MES"]
    # Second resolve should hit cache (no AI, no new entry creation)
    bd2 = asyncio.run(resolver.resolve("MES", df4_unambiguous_bullish()))
    assert bd2.allowed_direction == "BUY"
    assert state.bias_cache.entries["MES"] is cached_entry, "BiasEntry replaced -> cache miss"
    assert len(ai.calls) == 0
    _ok("resolve: cache hit within TTL -> no recompute, BiasEntry identity preserved")


# ============================================================
# 7. Cache miss after TTL: recompute
# ============================================================

def test_cache_miss_after_ttl():
    state = fresh_state()
    ai = FakeAIClient()
    resolver = BiasResolver(ai_client=ai, state=state, ttl_seconds=3600)
    asyncio.run(resolver.resolve("MES", df4_unambiguous_bullish()))
    # Force entry to be 2h old
    old = (utc_now() - timedelta(hours=2)).isoformat()
    e = state.bias_cache.entries["MES"]
    state.bias_cache.entries["MES"] = BiasEntry(
        direction=e.direction, confidence=e.confidence,
        rationale=e.rationale, computed_at=old,
    )
    bd2 = asyncio.run(resolver.resolve("MES", df4_unambiguous_bullish()))
    assert bd2.allowed_direction == "BUY"
    new_entry = state.bias_cache.entries["MES"]
    assert new_entry.computed_at != old, "TTL expiry must trigger fresh computed_at"
    _ok("resolve: TTL expired -> recompute and refresh BiasEntry")


# ============================================================
# 8. invalidate(symbol) forces recompute
# ============================================================

def test_invalidate_symbol_forces_recompute():
    state = fresh_state()
    ai = FakeAIClient()
    resolver = BiasResolver(ai_client=ai, state=state)
    asyncio.run(resolver.resolve("MES", df4_unambiguous_bullish()))
    asyncio.run(resolver.resolve("MNQ", df4_unambiguous_bullish()))
    resolver.invalidate("MES")
    assert "MES" not in state.bias_cache.entries
    assert "MNQ" in state.bias_cache.entries
    asyncio.run(resolver.resolve("MES", df4_unambiguous_bullish()))
    assert "MES" in state.bias_cache.entries
    _ok("invalidate(symbol): drops only that symbol, recompute on next resolve")


# ============================================================
# 9. invalidate() clears all
# ============================================================

def test_invalidate_all_clears_cache():
    state = fresh_state()
    ai = FakeAIClient()
    resolver = BiasResolver(ai_client=ai, state=state)
    asyncio.run(resolver.resolve("MES", df4_unambiguous_bullish()))
    asyncio.run(resolver.resolve("MNQ", df4_unambiguous_bullish()))
    assert len(state.bias_cache.entries) == 2
    resolver.invalidate()
    assert state.bias_cache.entries == {}
    _ok("invalidate() with no arg -> entire cache cleared")


# ============================================================
# 10. AI returns garbage bias -> fallback algo
# ============================================================

def test_ai_garbage_bias_fallback():
    state = fresh_state()
    ai = FakeAIClient(canned=make_ai_response({
        "bias": "GARBAGE",
        "allowed_direction": "BUY",
        "h1_compatibility": 0.9,
        "reason": "noise",
    }))
    resolver = BiasResolver(ai_client=ai, state=state)
    bd = asyncio.run(resolver.resolve("MES", df4_ambiguous()))
    # V16 stricter than V15: garbage bias -> entire fallback to algo
    assert bd.bias == "NEUTRO", f"got {bd.bias}"
    assert not bd.h1_reason.startswith("AI: ")
    _ok("resolve: AI garbage bias -> fallback algo (V16 stricter than V15 silent-coerce)")


# ============================================================
# 11. NEUTRO + allowed=BOTH -> NONE (V14 coherence rule)
# ============================================================

def test_ai_neutro_with_both_coerced_to_none():
    """
    V15 bias_computer.py:254-259:
      - allowed not in {BUY,SELL,NONE} -> coerced to NONE
      - bias=NEUTRO -> SEMPRE allowed=NONE
    V15 prompt didn't even list BOTH (line 204), but defensive coercion
    handles a hallucinated AI response.
    """
    state = fresh_state()
    ai = FakeAIClient(canned=make_ai_response({
        "bias": "NEUTRO",
        "allowed_direction": "BOTH",
        "h1_compatibility": 0.6,
        "reason": "uncertain",
    }))
    resolver = BiasResolver(ai_client=ai, state=state)
    bd = asyncio.run(resolver.resolve("MES", df4_ambiguous()))
    assert bd.bias == "NEUTRO"
    assert bd.allowed_direction == "NONE", \
        f"V14 coherence: NEUTRO must force NONE, got {bd.allowed_direction}"
    _ok("resolve: AI NEUTRO+BOTH -> allowed coerced to NONE (V14 coherence)")


# ============================================================
# 12. h1_compatibility clamped to [0.3, 1.0]
# ============================================================

def test_ai_h1_compatibility_clamped():
    # Above ceiling
    state = fresh_state()
    ai = FakeAIClient(canned=make_ai_response({
        "bias": "RIALZISTA", "allowed_direction": "BUY",
        "h1_compatibility": 2.5, "reason": "high",
    }))
    bd = asyncio.run(BiasResolver(ai_client=ai, state=state).resolve("MES", df4_ambiguous()))
    assert bd.h1_compatibility == 1.0, f"got {bd.h1_compatibility}"

    # Below floor
    state2 = fresh_state()
    ai2 = FakeAIClient(canned=make_ai_response({
        "bias": "RIALZISTA", "allowed_direction": "BUY",
        "h1_compatibility": 0.1, "reason": "low",
    }))
    bd2 = asyncio.run(BiasResolver(ai_client=ai2, state=state2).resolve("MES", df4_ambiguous()))
    assert bd2.h1_compatibility == 0.3, f"got {bd2.h1_compatibility}"
    _ok("resolve: h1_compatibility 2.5 -> 1.0, 0.1 -> 0.3 (clamp [0.3, 1.0])")


# ============================================================
# 13. BiasData -> BiasEntry -> BiasData roundtrip on persisted fields
# ============================================================

def test_bias_data_to_entry_roundtrip():
    bd = BiasData(
        bias="RIALZISTA", allowed_direction="BUY",
        h1_compatibility=0.8, h1_reason="AI: bullish continuation",
        ambiguous=False, rsi_h4_override=False,
    )
    e = _bias_data_to_entry(bd)
    assert e.direction == "BUY"
    assert e.confidence == 80
    assert e.rationale == "AI: bullish continuation"
    bd2 = _entry_to_bias_data(e)
    assert bd2.allowed_direction == "BUY"
    assert bd2.bias == "RIALZISTA"        # derived from BUY
    assert bd2.h1_compatibility == 0.8
    assert bd2.h1_reason == "AI: bullish continuation"
    # Process flags reset on read
    assert bd2.ambiguous is False
    assert bd2.rsi_h4_override is False
    _ok("persist: BiasData<->BiasEntry roundtrip preserves persisted fields, resets flags")


# ============================================================
# 14. df4 None / short -> V15 router parity NEUTRO BOTH 0.5, cached
# ============================================================

def test_no_h4_data_returns_neutro_both():
    state = fresh_state()
    ai = FakeAIClient()
    resolver = BiasResolver(ai_client=ai, state=state)

    bd = asyncio.run(resolver.resolve("MES", None))
    assert bd.bias == "NEUTRO"
    assert bd.allowed_direction == "BOTH", \
        f"V15 router parity: no_h4_data -> BOTH (permissive default), got {bd.allowed_direction}"
    assert bd.h1_compatibility == 0.5
    assert bd.h1_reason == "no_h4_data"
    assert len(ai.calls) == 0
    # Cached for full TTL (V15 parity)
    assert "MES" in state.bias_cache.entries

    # Short df4 (only 5 rows) -> same path
    short = pd.DataFrame([{"open": 1, "high": 1, "low": 1, "close": 1, "volume": 0}] * 5)
    bd2 = asyncio.run(resolver.resolve("MNQ", short))
    assert bd2.allowed_direction == "BOTH"
    _ok("resolve: df4 None/short -> NEUTRO BOTH 0.5, cached, no AI")


# ============================================================
# MAIN
# ============================================================

def main() -> int:
    print("test_bias_resolver.py")
    test_resolve_unambiguous_skips_ai()
    test_resolve_ambiguous_calls_ai_happy()
    test_resolve_ambiguous_ai_parse_error_fallback()
    test_resolve_ambiguous_ai_timeout_fallback()
    test_resolve_ambiguous_ai_credit_fallback()
    test_cache_hit_within_ttl()
    test_cache_miss_after_ttl()
    test_invalidate_symbol_forces_recompute()
    test_invalidate_all_clears_cache()
    test_ai_garbage_bias_fallback()
    test_ai_neutro_with_both_coerced_to_none()
    test_ai_h1_compatibility_clamped()
    test_bias_data_to_entry_roundtrip()
    test_no_h4_data_returns_neutro_both()
    print("ALL 14 TESTS PASSED")
    return 0


if __name__ == "__main__":
    sys.exit(main())