trumpsignal-backend/tests/test_x_analysis.py

"""Tests for x_analysis — the DETERMINISTIC normalization layer.

The LLM call is mocked so each test feeds a controlled raw model response and
asserts how analyze_x_post normalizes it. This locks down the enforcement rules
that protect downstream consumers (kol_x → kol_divergence), independent of
whatever the model actually returns:

  - retweet post_type → forced noise
  - trade_signal requires a buy/sell/reduce ticker with conviction ≥ 0.7,
    else it downgrades to directional (never silently dropped)
  - noise → tickers cleared + talks_vs_trades_flag forced false
  - ticker hygiene: conviction clamped 0..1, bad action → mention,
    overlong symbol dropped
  - invalid tier/post_type → safe defaults
  - bad JSON / empty text → graceful fallback, never raises

These are pure logic (no network, no AI spend) so they're fast and stable.
"""

from __future__ import annotations

import json

import pytest

from app.config import settings
from app.services import x_analysis


# ── helpers ────────────────────────────────────────────────────────────────

def _patch_llm(monkeypatch, raw: dict):
    """Force analyze_x_post's LLM call to return `raw` (serialized). We patch
    the OpenAI path (anthropic key blanked) since that's the default in CI."""
    payload = json.dumps(raw)

    class _Msg:
        content = payload

    class _Choice:
        message = _Msg()

    class _Resp:
        choices = [_Choice()]

    class _Completions:
        async def create(self, **_kw):
            return _Resp()

    class _Chat:
        completions = _Completions()

    class _Client:
        chat = _Chat()

    monkeypatch.setattr(settings, "anthropic_api_key", "")   # → OpenAI path
    monkeypatch.setattr(x_analysis, "_oai", lambda: _Client())


def _raw(**over):
    """Minimal well-formed raw model response; override per test."""
    base = {
        "post_type": "original",
        "tier": "directional",
        "summary": "x",
        "tickers": [],
        "talks_vs_trades_flag": False,
        "has_price_target": False,
        "price_targets": [],
        "sentiment": "neutral",
        "reasoning": "y",
    }
    base.update(over)
    return base


# ── tier enforcement ───────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_retweet_post_type_forced_to_noise(monkeypatch):
    _patch_llm(monkeypatch, _raw(
        post_type="retweet", tier="trade_signal",
        tickers=[{"ticker": "BTC", "action": "buy", "conviction": 0.9}],
    ))
    r = await x_analysis.analyze_x_post(handle="k", text="some original-looking text body")
    assert r["tier"] == "noise"
    assert r["tickers"] == []


@pytest.mark.asyncio
async def test_trade_signal_downgrades_without_strong_action(monkeypatch):
    # tier=trade_signal but only a 'bullish' ticker (not buy/sell/reduce)
    _patch_llm(monkeypatch, _raw(
        tier="trade_signal",
        tickers=[{"ticker": "SOL", "action": "bullish", "conviction": 0.9}],
    ))
    r = await x_analysis.analyze_x_post(handle="k", text="SOL looking strong")
    assert r["tier"] == "directional"
    assert r["tickers"][0]["ticker"] == "SOL"


@pytest.mark.asyncio
async def test_trade_signal_downgrades_on_low_conviction(monkeypatch):
    _patch_llm(monkeypatch, _raw(
        tier="trade_signal",
        tickers=[{"ticker": "SOL", "action": "buy", "conviction": 0.5}],
    ))
    r = await x_analysis.analyze_x_post(handle="k", text="bought a little SOL maybe")
    assert r["tier"] == "directional"      # 0.5 < 0.7 floor


@pytest.mark.asyncio
async def test_trade_signal_kept_when_strong(monkeypatch):
    _patch_llm(monkeypatch, _raw(
        tier="trade_signal",
        tickers=[{"ticker": "SOL", "action": "buy", "conviction": 0.9}],
    ))
    r = await x_analysis.analyze_x_post(handle="k", text="aped SOL full size lfg")
    assert r["tier"] == "trade_signal"


@pytest.mark.asyncio
async def test_invalid_tier_falls_back_to_noise(monkeypatch):
    _patch_llm(monkeypatch, _raw(tier="超级买入"))
    r = await x_analysis.analyze_x_post(handle="k", text="ambiguous content here")
    assert r["tier"] == "noise"


# ── noise enforcement ──────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_noise_clears_tickers_and_flag(monkeypatch):
    _patch_llm(monkeypatch, _raw(
        tier="noise", talks_vs_trades_flag=True,
        tickers=[{"ticker": "BTC", "action": "buy", "conviction": 0.9}],
    ))
    r = await x_analysis.analyze_x_post(handle="k", text="gm frens beautiful day")
    assert r["tickers"] == []
    assert r["talks_vs_trades_flag"] is False


# ── ticker hygiene ─────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_conviction_clamped_to_unit_interval(monkeypatch):
    _patch_llm(monkeypatch, _raw(
        tier="directional",
        tickers=[{"ticker": "BTC", "action": "bullish", "conviction": 1.8}],
    ))
    r = await x_analysis.analyze_x_post(handle="k", text="BTC going parabolic")
    assert r["tickers"][0]["conviction"] == 1.0


@pytest.mark.asyncio
async def test_invalid_action_becomes_mention(monkeypatch):
    _patch_llm(monkeypatch, _raw(
        tier="directional",
        tickers=[{"ticker": "BTC", "action": "yolo", "conviction": 0.5}],
    ))
    r = await x_analysis.analyze_x_post(handle="k", text="BTC yolo time")
    assert r["tickers"][0]["action"] == "mention"


@pytest.mark.asyncio
async def test_overlong_ticker_dropped(monkeypatch):
    _patch_llm(monkeypatch, _raw(
        tier="directional",
        tickers=[{"ticker": "THISISWAYTOOLONG", "action": "bullish", "conviction": 0.5}],
    ))
    r = await x_analysis.analyze_x_post(handle="k", text="some long token mention")
    assert r["tickers"] == []


# ── graceful failure ───────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_empty_text_short_circuits_without_llm(monkeypatch):
    # no _patch_llm → if it called the LLM it would error; it must not.
    r = await x_analysis.analyze_x_post(handle="k", text="   ")
    assert r["tier"] == "noise"
    assert r["error"] == "empty post"


@pytest.mark.asyncio
async def test_bare_retweet_prefiltered_without_llm(monkeypatch):
    r = await x_analysis.analyze_x_post(handle="k", text="RT @someone: gm")
    assert r["tier"] == "noise"
    assert r["post_type"] == "retweet"


@pytest.mark.asyncio
async def test_bad_json_returns_fallback(monkeypatch):
    class _Msg:
        content = "not json at all {{{ "

    class _Choice:
        message = _Msg()

    class _Resp:
        choices = [_Choice()]

    class _Completions:
        async def create(self, **_kw):
            return _Resp()

    class _Chat:
        completions = _Completions()

    class _Client:
        chat = _Chat()

    monkeypatch.setattr(settings, "anthropic_api_key", "")
    monkeypatch.setattr(x_analysis, "_oai", lambda: _Client())

    r = await x_analysis.analyze_x_post(handle="k", text="real content that triggers llm")
    assert r["tier"] == "noise"
    assert r["error"] and "parse_error" in r["error"]