Multi_Swarm_Coevolutive/tests/unit/test_adversarial.py

import json

import numpy as np
import pandas as pd
import pytest

from multi_swarm.agents.adversarial import (
    AdversarialAgent,
    AdversarialReport,
    Severity,
)
from multi_swarm.backtest.engine import BacktestResult
from multi_swarm.backtest.orders import Side, Trade
from multi_swarm.protocol.parser import parse_strategy


@pytest.fixture
def ohlcv() -> pd.DataFrame:
    idx = pd.date_range("2024-01-01", periods=500, freq="1h", tz="UTC")
    close = 100 + np.cumsum(np.random.RandomState(0).normal(0.0, 1.0, 500))
    return pd.DataFrame(
        {
            "open": close,
            "high": close + 0.5,
            "low": close - 0.5,
            "close": close,
            "volume": 1.0,
        },
        index=idx,
    )


def test_degenerate_always_long_flagged(ohlcv: pd.DataFrame) -> None:
    src = json.dumps(
        {
            "rules": [
                {
                    "condition": {
                        "op": "gt",
                        "args": [
                            {"kind": "feature", "name": "close"},
                            {"kind": "literal", "value": -1e9},
                        ],
                    },
                    "action": "entry-long",
                }
            ]
        }
    )
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    assert isinstance(report, AdversarialReport)
    assert any(f.name == "degenerate" and f.severity == Severity.HIGH for f in report.findings)


def test_no_findings_on_reasonable_strategy(ohlcv: pd.DataFrame) -> None:
    src = json.dumps(
        {
            "rules": [
                {
                    "condition": {
                        "op": "gt",
                        "args": [
                            {"kind": "indicator", "name": "rsi", "params": [14]},
                            {"kind": "literal", "value": 70.0},
                        ],
                    },
                    "action": "entry-short",
                },
                {
                    "condition": {
                        "op": "lt",
                        "args": [
                            {"kind": "indicator", "name": "rsi", "params": [14]},
                            {"kind": "literal", "value": 30.0},
                        ],
                    },
                    "action": "entry-long",
                },
            ]
        }
    )
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    high_findings = [f for f in report.findings if f.severity == Severity.HIGH]
    assert len(high_findings) == 0


def test_zero_trade_strategy_flagged(ohlcv: pd.DataFrame) -> None:
    src = json.dumps(
        {
            "rules": [
                {
                    "condition": {
                        "op": "gt",
                        "args": [
                            {"kind": "feature", "name": "close"},
                            {"kind": "literal", "value": 1e9},
                        ],
                    },
                    "action": "entry-long",
                }
            ]
        }
    )
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    assert any(f.name == "no_trades" for f in report.findings)


# AST minimale valido (parser-acceptable). Usato nei test che monkeypatchano
# compile_strategy/BacktestEngine.run: il contenuto della strategia e'
# irrilevante perche' il signal/result viene iniettato.
_MINIMAL_STRATEGY_SRC = json.dumps(
    {
        "rules": [
            {
                "condition": {
                    "op": "gt",
                    "args": [
                        {"kind": "feature", "name": "close"},
                        {"kind": "literal", "value": 0.0},
                    ],
                },
                "action": "entry-long",
            }
        ]
    }
)


def _make_trade(
    entry_ts: pd.Timestamp,
    exit_ts: pd.Timestamp,
    entry_price: float,
    exit_price: float,
    side: Side = Side.LONG,
    fees_bp: float = 5.0,
) -> Trade:
    return Trade(
        entry_ts=entry_ts.to_pydatetime() if hasattr(entry_ts, "to_pydatetime") else entry_ts,
        exit_ts=exit_ts.to_pydatetime() if hasattr(exit_ts, "to_pydatetime") else exit_ts,
        side=side,
        size=1.0,
        entry_price=entry_price,
        exit_price=exit_price,
        fees_bp=fees_bp,
    )


def test_undertrading_under_10_is_high(monkeypatch: pytest.MonkeyPatch,
                                        ohlcv: pd.DataFrame) -> None:
    """5 trade su 500 bar -> HIGH undertrading (Phase 1.5: era MEDIUM <5)."""
    fake_trades = [
        _make_trade(
            ohlcv.index[i * 50],
            ohlcv.index[i * 50 + 10],
            entry_price=100.0,
            exit_price=101.0,
        )
        for i in range(5)
    ]
    fake_signals = pd.Series(
        [Side.LONG] * 250 + [Side.FLAT] * 250, index=ohlcv.index, dtype=object
    )

    def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult:  # type: ignore[no-untyped-def]
        return BacktestResult(
            equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
            returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
            trades=fake_trades,
        )

    def fake_compile(strategy):  # type: ignore[no-untyped-def]
        return lambda df: fake_signals

    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
    )
    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.compile_strategy", fake_compile
    )

    src = _MINIMAL_STRATEGY_SRC
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    assert any(
        f.name == "undertrading" and f.severity == Severity.HIGH
        for f in report.findings
    )


def test_overtrading_with_tighter_threshold(monkeypatch: pytest.MonkeyPatch,
                                             ohlcv: pd.DataFrame) -> None:
    """n_trades > n_bars/20 -> MEDIUM overtrading (Phase 1.5: era /5)."""
    # 500 bar / 20 = 25. Forziamo 30 trade.
    n = 30
    fake_trades = [
        _make_trade(
            ohlcv.index[i * 10],
            ohlcv.index[i * 10 + 5],
            entry_price=100.0,
            exit_price=100.5,
        )
        for i in range(n)
    ]
    # Signal alternato per evitare flat_too_long: 50% LONG, 50% FLAT.
    fake_signals = pd.Series(
        [Side.LONG if i % 2 == 0 else Side.FLAT for i in range(len(ohlcv))],
        index=ohlcv.index,
        dtype=object,
    )

    def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult:  # type: ignore[no-untyped-def]
        return BacktestResult(
            equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
            returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
            trades=fake_trades,
        )

    def fake_compile(strategy):  # type: ignore[no-untyped-def]
        return lambda df: fake_signals

    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
    )
    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.compile_strategy", fake_compile
    )

    src = _MINIMAL_STRATEGY_SRC
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    assert any(
        f.name == "overtrading" and f.severity == Severity.MEDIUM
        for f in report.findings
    )


def test_flat_too_long_flagged(monkeypatch: pytest.MonkeyPatch,
                                ohlcv: pd.DataFrame) -> None:
    """Signal flat per >95% delle bar -> HIGH flat_too_long."""
    n_bars = len(ohlcv)
    # 96% flat: 480 FLAT + 20 LONG = 96% flat ratio
    n_active = 20
    sig_values = [Side.LONG] * n_active + [Side.FLAT] * (n_bars - n_active)
    fake_signals = pd.Series(sig_values, index=ohlcv.index, dtype=object)
    # 15 trade per evitare undertrading HIGH.
    fake_trades = [
        _make_trade(
            ohlcv.index[i * 30],
            ohlcv.index[i * 30 + 1],
            entry_price=100.0,
            exit_price=101.0,
        )
        for i in range(15)
    ]

    def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult:  # type: ignore[no-untyped-def]
        return BacktestResult(
            equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
            returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
            trades=fake_trades,
        )

    def fake_compile(strategy):  # type: ignore[no-untyped-def]
        return lambda df: fake_signals

    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
    )
    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.compile_strategy", fake_compile
    )

    src = _MINIMAL_STRATEGY_SRC
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    assert any(
        f.name == "flat_too_long" and f.severity == Severity.HIGH
        for f in report.findings
    )


def test_fees_eat_alpha_flagged(monkeypatch: pytest.MonkeyPatch,
                                 ohlcv: pd.DataFrame) -> None:
    """gross_pnl > 0 ma fees > 50% del lordo -> HIGH fees_eat_alpha."""
    # Costruisco trade con gross piccolo e fees alti via fees_bp esagerato.
    # entry=100, exit=100.05, size=1 -> gross=0.05
    # fees_bp=200 (2%) su (100+100.05)*1*200/10000 = 4.001 fees per trade
    # In aggregato: gross=15*0.05=0.75, fees=15*4.001=60 -> ratio enorme.
    n = 15
    fake_trades = [
        _make_trade(
            ohlcv.index[i * 30],
            ohlcv.index[i * 30 + 1],
            entry_price=100.0,
            exit_price=100.05,
            fees_bp=200.0,
        )
        for i in range(n)
    ]
    # Signal misto per evitare flat_too_long. 50% attivo.
    fake_signals = pd.Series(
        [Side.LONG if i % 2 == 0 else Side.FLAT for i in range(len(ohlcv))],
        index=ohlcv.index,
        dtype=object,
    )

    def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult:  # type: ignore[no-untyped-def]
        return BacktestResult(
            equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
            returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
            trades=fake_trades,
        )

    def fake_compile(strategy):  # type: ignore[no-untyped-def]
        return lambda df: fake_signals

    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
    )
    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.compile_strategy", fake_compile
    )

    src = _MINIMAL_STRATEGY_SRC
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    assert any(
        f.name == "fees_eat_alpha" and f.severity == Severity.HIGH
        for f in report.findings
    )


def test_time_in_market_too_high_flagged(monkeypatch: pytest.MonkeyPatch,
                                          ohlcv: pd.DataFrame) -> None:
    """Signal LONG per >80% delle bar -> HIGH time_in_market_too_high."""
    n_bars = len(ohlcv)
    # 90% LONG, 10% FLAT iniziali (warmup-like) per evitare degenerate.
    n_flat = int(n_bars * 0.10)
    sig_values = [Side.FLAT] * n_flat + [Side.LONG] * (n_bars - n_flat)
    fake_signals = pd.Series(sig_values, index=ohlcv.index, dtype=object)
    # 15 trade per evitare undertrading HIGH.
    fake_trades = [
        _make_trade(
            ohlcv.index[i * 30],
            ohlcv.index[i * 30 + 1],
            entry_price=100.0,
            exit_price=101.0,
        )
        for i in range(15)
    ]

    def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult:  # type: ignore[no-untyped-def]
        return BacktestResult(
            equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
            returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
            trades=fake_trades,
        )

    def fake_compile(strategy):  # type: ignore[no-untyped-def]
        return lambda df: fake_signals

    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
    )
    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.compile_strategy", fake_compile
    )

    src = _MINIMAL_STRATEGY_SRC
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    assert any(
        f.name == "time_in_market_too_high" and f.severity == Severity.HIGH
        for f in report.findings
    )


def test_reasonable_balanced_strategy_not_flagged(monkeypatch: pytest.MonkeyPatch,
                                                    ohlcv: pd.DataFrame) -> None:
    """Mix ~50% flat, ~25% long, ~25% short: no HIGH sui gate temporali."""
    n_bars = len(ohlcv)
    # Pattern ciclico: 2 flat, 1 long, 1 short per ogni gruppo da 4 bar.
    # Risultato: ~50% FLAT, ~25% LONG, ~25% SHORT. flat_ratio=0.5 < 0.95,
    # active_ratio=0.5 < 0.80.
    pattern = [Side.FLAT, Side.FLAT, Side.LONG, Side.SHORT]
    sig_values = [pattern[i % 4] for i in range(n_bars)]
    fake_signals = pd.Series(sig_values, index=ohlcv.index, dtype=object)
    # 15 trade per evitare undertrading HIGH.
    fake_trades = [
        _make_trade(
            ohlcv.index[i * 30],
            ohlcv.index[i * 30 + 1],
            entry_price=100.0,
            exit_price=101.0,
        )
        for i in range(15)
    ]

    def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult:  # type: ignore[no-untyped-def]
        return BacktestResult(
            equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
            returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
            trades=fake_trades,
        )

    def fake_compile(strategy):  # type: ignore[no-untyped-def]
        return lambda df: fake_signals

    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
    )
    monkeypatch.setattr(
        "multi_swarm.agents.adversarial.compile_strategy", fake_compile
    )

    src = _MINIMAL_STRATEGY_SRC
    ast = parse_strategy(src)
    agent = AdversarialAgent()
    report = agent.review(ast, ohlcv)
    # I due gate temporali non devono triggerare.
    names = [f.name for f in report.findings]
    assert "flat_too_long" not in names
    assert "time_in_market_too_high" not in names