trumpsignal-backend/tests/test_bottom_indicators_and_leverage.py

"""Tests for the current BTC bottom strategy:

  - pure-price indicators (AHR999 / 200WMA / Pi Cycle) + 2-of-3 confluence
  - leverage-aware dynamic System-2 leverage + protective stop
  - staged stop-loss ladder math (no take-profit, ratchet-up only)
"""

import math

from app.services.bottom_indicators import (
    ahr999, below_200wma, pi_cycle_bottom, bottom_confluence,
    AHR999_BOTTOM,
)
from app.services.signal_categories import (
    sys2_effective_leverage, sys2_protective_stop_pct,
    sys2_approx_liquidation_pct, get_stop_ladder, sys2_derisk_ladder,
    sys2_addon_ladder, sys2_peak_trail, get_exit_profile,
    SYS2_DEFAULT_LEVERAGE, SYS2_MAX_STOP_PCT,
)
from app.services.bottom_indicators import trend_confirmed


# ── Indicators ──────────────────────────────────────────────────────────────

def test_ahr999_none_when_insufficient_history():
    assert ahr999([100.0] * 199) is None


def test_ahr999_low_in_deep_decline():
    # Long steady decline → price far below the geometric mean & growth model.
    daily = [60000 * math.exp(-0.002 * i) for i in range(400)]
    v = ahr999(daily)
    assert v is not None and v < AHR999_BOTTOM


def test_below_200wma_uses_explicit_price_not_stale_weekly():
    weekly = [100.0] * 199 + [130.0]   # 200-wk mean ≈ 100.15, last close 130
    # Using the stale weekly close → above band → False.
    sig_stale, wma = below_200wma(weekly)
    assert sig_stale is False and wma is not None
    # Passing the fresh (lower) daily price → below band → True.
    sig_fresh, _ = below_200wma(weekly, price=95.0)
    assert sig_fresh is True


def test_pi_cycle_bottom_none_when_short():
    assert pi_cycle_bottom([1.0] * 470)[0] is False


def test_confluence_fires_only_with_two_of_three():
    # Deep decline: AHR999 deep-value + 200WMA both true → ≥2 → fired.
    daily = [60000 * math.exp(-0.002 * i) for i in range(520)]
    weekly = [60000 * math.exp(-0.01 * i) for i in range(210)]
    c = bottom_confluence(daily, weekly)
    assert c.votes >= 2 and c.fired is True
    # Strong bull: AHR999 expensive (>0.45) and Pi not in bottom region →
    # at most 1 vote → confluence requires ≥2 so it must NOT fire.
    up_d = [20000 * math.exp(0.004 * i) for i in range(520)]
    up_w = [20000 * math.exp(0.02 * i) for i in range(210)]
    c2 = bottom_confluence(up_d, up_w)
    assert c2.detail["signals"]["ahr999_value"] is False
    assert c2.detail["signals"]["pi_cycle_bottom"] is False
    assert c2.votes < 2 and c2.fired is False


def test_confluence_b_uses_latest_daily_price():
    # weekly close stale-high but the latest daily close is at the lows.
    daily = [60000 * math.exp(-0.002 * i) for i in range(520)]
    weekly = [30000.0] * 209 + [90000.0]   # last weekly close artificially high
    c = bottom_confluence(daily, weekly)
    # B must compare the 200wma against the latest DAILY close, not 90000.
    assert c.detail["price"] == round(daily[-1], 2)


# ── Dynamic System-2 leverage ────────────────────────────────────────────────

def test_effective_leverage_clamps_and_defaults():
    assert sys2_effective_leverage(None) == SYS2_DEFAULT_LEVERAGE
    assert sys2_effective_leverage(0) == 1
    assert sys2_effective_leverage(99) == 10
    assert sys2_effective_leverage(5) == 5
    assert sys2_effective_leverage("bad") == SYS2_DEFAULT_LEVERAGE


def test_protective_stop_always_inside_liquidation():
    # The protective full-exit must trigger BEFORE the exchange liquidates,
    # for every allowed leverage.
    for lev in range(1, 11):
        stop = sys2_protective_stop_pct(lev)
        liq = sys2_approx_liquidation_pct(lev)
        assert stop < liq, f"lev {lev}: stop {stop} not inside liq {liq}"
    # Low leverage keeps the full bottom-wick tolerance.
    assert sys2_protective_stop_pct(1) == SYS2_MAX_STOP_PCT
    assert sys2_protective_stop_pct(2) == SYS2_MAX_STOP_PCT
    # High leverage tightens automatically.
    assert sys2_protective_stop_pct(5) < 20
    assert sys2_protective_stop_pct(10) < 10


def test_stop_ladder_has_no_negative_base_rung_and_only_ratchets_up():
    ladder = get_stop_ladder("btc_bottom_reversal_long")
    assert ladder is not None
    triggers = [t for t, _ in ladder]
    floors = [f for _, f in ladder]
    # Sorted ascending by trigger.
    assert triggers == sorted(triggers)
    # Floors strictly increase (pure ratchet-up: never loosens).
    assert floors == sorted(floors)
    # No 0%-trigger catastrophic rung — that floor is leverage-derived now.
    assert triggers[0] > 0
    # Reaches locked-in profit (a positive floor exists).
    assert max(floors) > 0


def test_non_bottom_category_has_no_ladder():
    assert get_stop_ladder("sma_reclaim") is None
    assert get_stop_ladder(None) is None


# ── Staged-stop monitor math (mirror of tp_sl_monitor branch 0) ──────────────

def _eff_stop(peak: float, base_stop_pct: float, ladder) -> float:
    eff = -base_stop_pct
    for trig, floor in ladder:
        if peak >= trig and floor > eff:
            eff = floor
    return eff


def test_staged_stop_no_take_profit_and_locks_profit():
    ladder = get_stop_ladder("btc_bottom_reversal_long")
    base = sys2_protective_stop_pct(2)   # 35%

    # Never exits for a big unrealised gain (no take-profit).
    assert 300.0 > _eff_stop(peak=300.0, base_stop_pct=base, ladder=ladder)

    # Catastrophic floor before any rung: -35% at 2x.
    assert _eff_stop(peak=0.0, base_stop_pct=base, ladder=ladder) == -35.0

    # After +70% peak the floor has ratcheted to a locked-in profit.
    locked = _eff_stop(peak=75.0, base_stop_pct=base, ladder=ladder)
    assert locked > 0

    # High leverage → tighter base floor, ladder still ratchets identically.
    base10 = sys2_protective_stop_pct(10)
    assert _eff_stop(peak=0.0, base_stop_pct=base10, ladder=ladder) == -base10
    assert _eff_stop(peak=75.0, base_stop_pct=base10, ladder=ladder) == locked


# ── Staged de-risk ladder (分段式减仓) ───────────────────────────────────────

def test_derisk_ladder_shape_and_safety():
    for lev in (1, 2, 3, 5, 10):
        p = sys2_protective_stop_pct(lev)
        liq = sys2_approx_liquidation_pct(lev)
        ladder = sys2_derisk_ladder(lev)
        assert len(ladder) == 3
        thrs = [t for t, _, _ in ladder]
        fracs = [f for _, f, _ in ladder]
        finals = [fin for _, _, fin in ladder]
        # All thresholds negative, increasing in adversity.
        assert all(t < 0 for t in thrs)
        assert thrs == sorted(thrs, reverse=True)   # -21, -28, -35 …
        # Exactly the last rung is the full close.
        assert finals == [False, False, True]
        # Fractions sum to the whole position (thirds of original).
        assert abs(sum(fracs) - 1.0) < 1e-9
        # Final rung == the protective level == inside liquidation.
        assert abs(thrs[-1] - (-p)) < 1e-6
        assert -thrs[-1] < liq, f"lev {lev}: final {thrs[-1]} not inside liq {liq}"


def test_derisk_pnl_accounting_matches_single_close():
    """Summing the staged slices + the remaining close must equal a single
    full close at the same final price (the staged path must not create or
    destroy PnL vs closing all at once)."""
    notional = 1000.0
    entry = 100.0
    final_price = 70.0   # -30% (long)

    def slice_pnl(frac, px):
        return notional * frac * ((px - entry) / entry)

    # Staged: 1/3 closed at 85, 1/3 at 78, final 1/3 at 70.
    staged = (
        slice_pnl(1/3, 85.0) +
        slice_pnl(1/3, 78.0) +
        slice_pnl(1/3, 70.0)
    )
    # If instead all three thirds were closed at the final price:
    single = slice_pnl(1.0, final_price)
    # Staged exits EARLIER on the way down, so it must lose LESS than a
    # single close at the worst price (the whole point of de-risking).
    assert staged > single
    # And closing all at entry-price slices would be zero — sanity.
    assert abs(slice_pnl(1/3, entry) * 3) < 1e-9


def test_derisk_regime_switch_underwater_vs_profit():
    """Mirror the monitor's regime decision: underwater → de-risk ladder,
    in-profit (peak ≥ first upside rung) → ratchet stop."""
    stop_ladder = get_stop_ladder("btc_bottom_reversal_long")
    first_up = min(t for t, _ in stop_ladder)   # 20

    def regime(peak):
        return "profit" if peak >= first_up else "derisk"

    assert regime(0.0) == "derisk"
    assert regime(19.9) == "derisk"
    assert regime(20.0) == "profit"
    assert regime(150.0) == "profit"


# ── Pyramiding (做对了往上加仓) ──────────────────────────────────────────────

def test_addon_ladder_shape_conservative():
    ladder = sys2_addon_ladder()
    trigs = [t for t, _, _ in ladder]
    fracs = [f for _, f, _ in ladder]
    lasts = [x for _, _, x in ladder]
    assert trigs == sorted(trigs) and all(t > 0 for t in trigs)   # peak-gain rungs
    assert fracs[:3] == [0.30, 0.20, 0.10]                        # conservative base
    assert sum(fracs) <= 0.80                                     # still modest total
    assert lasts[-1] is True and lasts.count(True) == 1           # exactly one final


def test_trend_confirmed_requires_sma_and_new_high():
    # Uptrend: price above 200d SMA and at a fresh 20d high → confirmed.
    up = [100.0 + i for i in range(260)]
    highs = [c + 1 for c in up]
    price = up[-1] + 1
    assert trend_confirmed(up, highs, price) is True
    # Below the 200d SMA → not confirmed even at a local high.
    assert trend_confirmed(up, highs, price=50.0) is False
    # Above SMA but well below the recent high (still chopping) → not confirmed.
    mid = sum(up[-200:]) / 200
    assert trend_confirmed(up, highs, price=mid + 1) is False
    # Too little history → fail closed.
    assert trend_confirmed([1.0] * 50, [1.0] * 50, 1.0) is False


def test_pyramiding_blended_entry_math():
    """Blended entry after an add must be the notional-weighted average, and
    it must sit BELOW the add price for a long that added higher (so the
    aggregate is still in profit)."""
    base = 1000.0
    entry = 100.0
    add_usd = base * 0.30
    fill = 130.0                       # added after +30%
    new_notional = base + add_usd
    blended = (base * entry + add_usd * fill) / new_notional
    assert entry < blended < fill      # weighted average
    # Aggregate still profitable at the add price.
    assert (fill - blended) / blended > 0
    # Conservative sizing only modestly lifts the average (< 7% here).
    assert (blended - entry) / entry < 0.07


def test_recovery_restores_peak_keeps_profit_regime():
    """A pyramided / in-profit trade rehydrated after a restart must seed the
    monitor's peak from the persisted value so it stays in the profit regime
    (not fall back to the underwater de-risk regime)."""
    from app.services.tp_sl_monitor import register_trade, _watched, unregister
    from app.services.signal_categories import (
        sys2_derisk_ladder, sys2_addon_ladder, get_stop_ladder,
    )
    tid = 990011
    try:
        register_trade(
            trade_id=tid, wallet="0xabc", api_key="k", leverage=2,
            asset="BTC", side="long", entry_price=100.0,
            take_profit_pct=None, stop_loss_pct=35.0,
            stop_ladder=get_stop_ladder("btc_bottom_reversal_long"),
            derisk_ladder=sys2_derisk_ladder(2),
            addon_ladder=sys2_addon_ladder(), addon_done=1,
            initial_peak=90.0,
        )
        wt = _watched[tid]
        assert wt.peak_gain_pct == 90.0
        assert wt.peak_persisted == 90.0
        first_up = min(t for t, _ in wt.stop_ladder)
        # peak 90 ≥ first upside rung → profit regime, NOT underwater de-risk.
        assert wt.peak_gain_pct >= first_up
    finally:
        unregister(tid)


def test_addon_ladder_extended_for_cycle_bull():
    ladder = sys2_addon_ladder()
    trigs = [t for t, _, _ in ladder]
    fracs = [f for _, f, _ in ladder]
    lasts = [x for _, _, x in ladder]
    assert len(ladder) == 5                       # deeper continuation rungs
    assert trigs == sorted(trigs) and trigs[-1] >= 200
    assert lasts == [False, False, False, False, True]
    assert abs(sum(fracs) - 0.75) < 1e-9          # ≤ +0.75× base, still modest


def test_btc_bottom_maxhold_is_18_months():
    p = get_exit_profile("btc_bottom_reversal_long")
    assert p.max_hold_hours == 12960              # 540 days ≈ 18 months


def _peak_trail_floor(peak_pp: float):
    start, dd = sys2_peak_trail()
    if peak_pp < start:
        return None
    return ((1.0 + peak_pp / 100.0) * (1.0 - dd) - 1.0) * 100.0


def test_peak_trail_scale_invariant_and_never_loosens():
    start, dd = sys2_peak_trail()
    # Inactive below the start threshold.
    assert _peak_trail_floor(start - 1) is None
    # Self-scales: a +500% move locks far above the old fixed +95% top rung.
    f500 = _peak_trail_floor(500.0)
    assert f500 is not None and f500 > 300.0      # ≈ +320%
    # A +900% move scales further still (not capped).
    assert _peak_trail_floor(900.0) > f500
    # Survives a normal bull pullback: at peak +120% the floor is a ≤30%
    # PRICE drawdown from the peak, i.e. price 2.2→1.54 (gain +54%), so a
    # typical 20–25% dip (still >+54%) does NOT trip it.
    f120 = _peak_trail_floor(120.0)
    assert 50.0 < f120 < 60.0                      # ((2.2*0.7)-1)*100 = 54
    # Combined with the fixed rungs the floor only ratchets UP (max of both).
    rung_at_160 = 95.0
    assert max(rung_at_160, _peak_trail_floor(160.0)) == rung_at_160
    assert max(95.0, _peak_trail_floor(400.0)) == _peak_trail_floor(400.0)


def test_sys2_risk_mode_params():
    from app.services.signal_categories import (
        sys2_normalize_mode, sys2_addon_ladder, sys2_derisk_ladder,
        sys2_peak_trail,
    )
    assert sys2_normalize_mode("AGGRESSIVE") == "aggressive"
    assert sys2_normalize_mode("garbage") == "standard"
    assert sys2_normalize_mode(None) == "standard"

    # Leverage default depends on mode; explicit value still clamped.
    assert sys2_effective_leverage(None, "standard") == 2
    assert sys2_effective_leverage(None, "aggressive") == 8
    assert sys2_effective_leverage(99, "aggressive") == 10

    # Aggressive pyramiding: earlier + heavier, ≤ +1.5× base, one final.
    ag = sys2_addon_ladder("aggressive")
    st = sys2_addon_ladder("standard")
    assert [t for t, _, _ in ag] == [15.0, 35.0, 60.0, 100.0, 160.0]
    assert abs(sum(f for _, f, _ in ag) - 1.50) < 1e-9
    assert [x for _, _, x in ag][-1] is True
    assert ag != st                                  # genuinely different
    assert abs(sum(f for _, f, _ in st) - 0.75) < 1e-9   # standard unchanged

    # Aggressive de-risk keeps a bigger runner (¼/¼/½) but final still FULL.
    agd = sys2_derisk_ladder(8, "aggressive")
    assert [round(f, 4) for _, f, _ in agd] == [0.25, 0.25, 0.5]
    assert agd[-1][2] is True                        # final = full close
    std_d = sys2_derisk_ladder(8, "standard")
    assert abs(std_d[0][1] - 1.0 / 3.0) < 1e-9       # standard unchanged
    assert std_d[-1][2] is True

    # Aggressive peak-trail: earlier start, wider give-back.
    assert sys2_peak_trail("aggressive") == (60.0, 0.42)
    assert sys2_peak_trail("standard") == (80.0, 0.30)
    assert sys2_peak_trail() == (80.0, 0.30)         # default = standard

    # Safety invariant holds in BOTH modes: final de-risk rung is the
    # protective level (inside liquidation) and is a full close.
    for m in ("standard", "aggressive"):
        for lev in (2, 5, 8, 10):
            lad = sys2_derisk_ladder(lev, m)
            assert lad[-1][2] is True
            assert -lad[-1][0] < sys2_approx_liquidation_pct(lev)


def test_register_trade_default_peak_is_zero():
    from app.services.tp_sl_monitor import register_trade, _watched, unregister
    tid = 990012
    try:
        register_trade(
            trade_id=tid, wallet="0xabc", api_key="k", leverage=3,
            asset="BTC", side="long", entry_price=100.0,
            take_profit_pct=None, stop_loss_pct=6.0,
        )
        assert _watched[tid].peak_gain_pct == 0.0
    finally:
        unregister(tid)


def test_register_trade_grow_mode_default_off_and_settable():
    """Per-trade Grow defaults OFF (pyramiding opt-in) and is settable."""
    from app.services.tp_sl_monitor import register_trade, _watched, unregister
    a, b = 990021, 990022
    try:
        register_trade(
            trade_id=a, wallet="0xabc", api_key="k", leverage=3,
            asset="BTC", side="long", entry_price=100.0,
            take_profit_pct=None, stop_loss_pct=6.0,
        )
        assert _watched[a].grow_mode is False          # default OFF
        register_trade(
            trade_id=b, wallet="0xabc", api_key="k", leverage=2,
            asset="BTC", side="long", entry_price=100.0,
            take_profit_pct=None, stop_loss_pct=35.0,
            grow_mode=True,
        )
        assert _watched[b].grow_mode is True
    finally:
        unregister(a); unregister(b)


def test_auto_trade_gate_skips_when_off():
    """The master gate: process_post must NOT open a trade when the sub has
    auto_trade falsy (the signal Post is still created upstream by ingest)."""
    import inspect
    from app.services import bot_engine
    # The per-subscriber executor holds the gate: keyed off auto_trade,
    # returns (no trade opened) when off.
    gate = inspect.getsource(bot_engine._execute_for_subscriber)
    assert 'sub.get("auto_trade")' in gate
    assert "Auto-Trade OFF" in gate
    assert "return" in gate.split('sub.get("auto_trade")')[1][:260]
    # process_post builds the snapshot carrying auto_trade so the gate reads it.
    pp = inspect.getsource(bot_engine.process_post)
    assert "auto_trade=bool(s.auto_trade)" in pp