k
170 lines
6.9 KiB
Python
170 lines
6.9 KiB
Python
"""Validation-path tests for the adopt / release flow.
|
||
|
||
These cover the in-memory branches that don't need a real DB / real HL —
|
||
mode normalisation, the error-code surface, and the structure of the
|
||
returned objects. End-to-end (real HL state → BotTrade row written →
|
||
watchdog registered) is verified by the manual /adopt Telegram flow,
|
||
not in CI.
|
||
|
||
Why so light: every interesting failure case is a DB or HL state issue
|
||
that's better caught manually. The function shape + error codes are what
|
||
the API + Telegram layer switch on, so those we lock down here.
|
||
"""
|
||
import pytest
|
||
|
||
from app.services.adoption import (
|
||
AdoptionError, ADOPTED_CATEGORY,
|
||
HLPositionView, AdoptionResult,
|
||
)
|
||
from app.services.signal_categories import (
|
||
sys2_normalize_mode, get_exit_profile, get_stop_ladder,
|
||
sys2_protective_stop_pct, sys2_derisk_ladder, sys2_addon_ladder,
|
||
sys2_peak_trail, SYS2_MODES,
|
||
)
|
||
|
||
|
||
def test_adopted_category_resolves_to_btc_bottom_profile():
|
||
"""The adopted-position category must point to a real exit profile —
|
||
otherwise adoption silently falls back to the System-2 default
|
||
(different stop / different max-hold)."""
|
||
profile = get_exit_profile(ADOPTED_CATEGORY)
|
||
assert profile.stop_ladder is not None, \
|
||
"adopted category must use the staged stop ladder"
|
||
assert profile.max_hold_hours == 12960, \
|
||
"BTC bottom-reversal is an 18-month-max-hold strategy"
|
||
# 5 upside rungs match the documented strategy.
|
||
ladder = get_stop_ladder(ADOPTED_CATEGORY)
|
||
assert len(ladder) == 5
|
||
|
||
|
||
def test_adoption_error_carries_code_and_message():
|
||
"""API + Telegram layers switch on .code; ensure both attrs are set."""
|
||
exc = AdoptionError("concurrency_cap", "You already have 3.")
|
||
assert exc.code == "concurrency_cap"
|
||
assert exc.message == "You already have 3."
|
||
assert "3" in str(exc)
|
||
|
||
|
||
def test_mode_normalisation_maps_unknown_to_standard():
|
||
assert sys2_normalize_mode("standard") == "standard"
|
||
assert sys2_normalize_mode("aggressive") == "aggressive"
|
||
assert sys2_normalize_mode("STANDARD") == "standard"
|
||
assert sys2_normalize_mode("") == "standard"
|
||
assert sys2_normalize_mode(None) == "standard"
|
||
assert sys2_normalize_mode("yolo") == "standard"
|
||
|
||
|
||
def test_mode_set_matches_canonical_list():
|
||
"""SYS2_MODES is the source of truth for what adopt_position accepts;
|
||
if we add a third mode this test fails on purpose to remind us to
|
||
update the bot's mode picker buttons too."""
|
||
assert SYS2_MODES == ("standard", "aggressive")
|
||
|
||
|
||
@pytest.mark.parametrize("lev,expected_max_prot", [
|
||
(1, 35.0), # 1× → 85% of 100% liq = 85%, capped at 35%
|
||
(2, 35.0), # 2× → 85% of 50% = 42.5%, capped at 35%
|
||
(3, 28.33), # 3× → 85% of 33.3% ≈ 28.33
|
||
(5, 17.0), # 5× → 85% of 20% = 17.0
|
||
(10, 8.5), # 10× → 85% of 10% = 8.5
|
||
])
|
||
def test_protective_stop_scales_with_leverage(lev, expected_max_prot):
|
||
"""The whole point of recomputing protective stop against HL's actual
|
||
leverage is that this number must stay INSIDE the liquidation distance.
|
||
A regression here is a real-money risk regression."""
|
||
got = sys2_protective_stop_pct(lev)
|
||
assert got == pytest.approx(expected_max_prot, abs=0.1)
|
||
|
||
|
||
def test_derisk_ladder_uses_three_rungs_summing_to_protective():
|
||
"""At any leverage, the three de-risk rungs should land at
|
||
-0.60·P, -0.80·P, -1.00·P. The final rung IS the full protective
|
||
close — same level as a one-shot stop, but staged."""
|
||
for lev in (2, 5, 10):
|
||
prot = sys2_protective_stop_pct(lev)
|
||
ladder = sys2_derisk_ladder(lev, "standard")
|
||
assert len(ladder) == 3
|
||
thrs = [r[0] for r in ladder]
|
||
assert thrs[0] == pytest.approx(-0.60 * prot, abs=0.5)
|
||
assert thrs[1] == pytest.approx(-0.80 * prot, abs=0.5)
|
||
assert thrs[2] == pytest.approx(-1.00 * prot, abs=0.5)
|
||
# Last rung is the FULL close (is_final flag).
|
||
assert ladder[-1][2] is True
|
||
|
||
|
||
def test_addon_ladder_differs_between_modes():
|
||
"""Aggressive mode must fire earlier and add more — otherwise the mode
|
||
label is a lie."""
|
||
std = sys2_addon_ladder("standard")
|
||
agg = sys2_addon_ladder("aggressive")
|
||
assert std[0][0] > agg[0][0], \
|
||
"aggressive first rung must trigger at a LOWER peak gain"
|
||
assert agg[0][1] > std[0][1], \
|
||
"aggressive first rung must add MORE notional"
|
||
|
||
|
||
def test_peak_trail_aggressive_more_permissive():
|
||
"""Aggressive peak-trail must let the trade run longer + give back
|
||
more drawdown — that's the explicit trade-off vs standard."""
|
||
std_start, std_dd = sys2_peak_trail("standard")
|
||
agg_start, agg_dd = sys2_peak_trail("aggressive")
|
||
assert agg_start < std_start
|
||
assert agg_dd > std_dd
|
||
|
||
|
||
def test_hl_position_view_dataclass_shape():
|
||
"""The API layer rehydrates HLPositionView via **vars(it); guard the
|
||
field set so a silent rename doesn't break the HTTP shape."""
|
||
v = HLPositionView(asset="BTC", side="long", size_coins=0.02,
|
||
entry_price=72000.0, leverage=2, size_usd=1440.0,
|
||
already_adopted=False)
|
||
d = vars(v)
|
||
assert set(d.keys()) == {
|
||
"asset", "side", "size_coins", "entry_price",
|
||
"leverage", "size_usd", "already_adopted",
|
||
}
|
||
|
||
|
||
def test_adoption_result_dataclass_shape():
|
||
r = AdoptionResult(
|
||
trade_id=1, asset="BTC", side="long", entry_price=72000.0,
|
||
size_usd=1440.0, leverage=2, mode="standard",
|
||
protective_stop=35.0, derisk_ladder=[], stop_ladder=[],
|
||
)
|
||
assert r.trade_id == 1
|
||
assert r.mode == "standard"
|
||
|
||
|
||
def test_wallet_adopt_lock_is_per_wallet_and_reused():
|
||
"""The lock cache must hand back the SAME lock for the same wallet
|
||
(otherwise two concurrent adopts would acquire different locks and
|
||
bypass serialisation), and a DIFFERENT lock for a different wallet."""
|
||
from app.services.adoption import _wallet_adopt_lock
|
||
a1 = _wallet_adopt_lock("0xaaa")
|
||
a2 = _wallet_adopt_lock("0xaaa")
|
||
b1 = _wallet_adopt_lock("0xbbb")
|
||
assert a1 is a2
|
||
assert a1 is not b1
|
||
|
||
|
||
def test_adoption_error_codes_used_by_callers_are_all_defined():
|
||
"""The API and Telegram layers switch on AdoptionError.code values.
|
||
If we ever silently rename one, the switch falls through to the
|
||
generic branch and the user loses the specific guidance. Lock the
|
||
set of codes we promise to emit."""
|
||
# These are the codes referenced from app/api/positions.py +
|
||
# app/services/telegram_bot.py. Update both sites if you add one.
|
||
promised_codes = {
|
||
"no_subscription", "no_hl_key", "paper_mode", "macro_disabled",
|
||
"key_decrypt_failed", "hl_read_failed",
|
||
"bad_mode", "circuit_breaker",
|
||
"already_adopted", "concurrency_cap",
|
||
"not_on_hl", "bad_hl_state",
|
||
"not_found", "not_owner", "already_closed",
|
||
}
|
||
# Just exercising the constructor for each — proves the codes
|
||
# round-trip cleanly through __init__.
|
||
for code in promised_codes:
|
||
err = AdoptionError(code, f"sample for {code}")
|
||
assert err.code == code
|