from __future__ import annotations import json from dataclasses import replace from decimal import Decimal from pathlib import Path import pytest from src.hyperliquid.market_data import L2BookSummary from src.market.trend_retest_features import TrendRetestFeatures, build_trend_retest_features from src.research.tradingview_context import TradingViewContext, classify_tradingview_impulse from src.strategies.trend_retest_anti_chase_v2 import LONG_STRATEGY_ID, SHORT_RESEARCH_ID, entry_blockers from src.strategies.relative_strength_momentum_v77_2 import MOMENTUM_RESEARCH_ID, MOMENTUM_VERSION from src.tools import v77_trend_retest_runtime as runtime def _candles(*, interval_ms: int, closes: list[Decimal], now_ms: int, latest: dict[str, str] | None = None) -> list[dict[str, str | int]]: rows = [] start = now_ms - len(closes) * interval_ms for idx, close in enumerate(closes): open_price = close - Decimal("0.05") row = { "t": start + idx * interval_ms, "T": start + (idx + 1) * interval_ms - 1, "o": str(open_price), "h": str(close + Decimal("0.20")), "l": str(open_price - Decimal("0.20")), "c": str(close), "v": "5000", } rows.append(row) if latest: rows[-1].update(latest) return rows def _feature(**changes) -> TrendRetestFeatures: base = TrendRetestFeatures( coin="BTC", current_price=Decimal("100"), candle_ts=200, data_window_id="window-1", open_15m=Decimal("99.5"), high_15m=Decimal("100.5"), low_15m=Decimal("98.8"), close_15m=Decimal("100"), move_15m_pct=Decimal("0.50"), move_1h_pct=Decimal("0.60"), sma_1h_fast=Decimal("99"), sma_1h_slow=Decimal("97"), sma_4h_fast=Decimal("96"), sma_4h_slow=Decimal("94"), atr_pct=Decimal("0.80"), rsi_15m=Decimal("60"), volume_24h=Decimal("50000000"), volume_ratio=Decimal("1.50"), recent_high=Decimal("101"), recent_low=Decimal("99"), pullback_from_high_pct=Decimal("0.99"), rebound_from_low_pct=Decimal("1.01"), spread_pct=Decimal("0.02"), bid_depth_notional_5=Decimal("50000"), ask_depth_notional_5=Decimal("50000"), buy_impact_1k_pct=Decimal("0.02"), sell_impact_1k_pct=Decimal("0.02"), funding_rate_hourly_pct=Decimal("0"), open_interest=Decimal("1000"), premium_pct=Decimal("0"), strength_rank=1, weakness_rank=4, breadth_positive_pct=Decimal("75"), ) return replace(base, **changes) def test_feature_builder_uses_closed_ohlcv_and_no_proxy_values(): now_ms = 2_000_000_000_000 c15 = _candles(interval_ms=900_000, closes=[Decimal("99") + Decimal(i) / Decimal("100") for i in range(60)], now_ms=now_ms, latest={"o": "99.4", "h": "100.4", "l": "98.8", "c": "100", "v": "10000"}) c1h = _candles(interval_ms=3_600_000, closes=[Decimal("90") + Decimal(i) / Decimal("5") for i in range(60)], now_ms=now_ms) c4h = _candles(interval_ms=14_400_000, closes=[Decimal("80") + Decimal(i) / Decimal("3") for i in range(60)], now_ms=now_ms) feature = build_trend_retest_features(coin="BTC", candles_15m=c15, candles_1h=c1h, candles_4h=c4h, now_ms=now_ms, spread_pct=Decimal("0.02"), bid_depth_notional_5=Decimal("50000"), ask_depth_notional_5=Decimal("50000"), buy_impact_1k_pct=Decimal("0.02"), sell_impact_1k_pct=Decimal("0.02")) assert feature.current_price == Decimal("100") assert feature.volume_ratio > Decimal("1.9") assert feature.atr_pct > 0 assert feature.sma_1h_fast != feature.current_price assert feature.data_window_id assert feature.data_quality_allowed is True def test_data_window_identity_changes_when_ohlcv_is_revised_at_same_timestamp(): now_ms = 2_000_000_000_000 c15 = _candles(interval_ms=900_000, closes=[Decimal("99") + Decimal(i) / Decimal("100") for i in range(60)], now_ms=now_ms) c1h = _candles(interval_ms=3_600_000, closes=[Decimal("90") + Decimal(i) / Decimal("5") for i in range(60)], now_ms=now_ms) c4h = _candles(interval_ms=14_400_000, closes=[Decimal("80") + Decimal(i) / Decimal("3") for i in range(60)], now_ms=now_ms) def build(candles_15m): return build_trend_retest_features(coin="BTC", candles_15m=candles_15m, candles_1h=c1h, candles_4h=c4h, now_ms=now_ms, spread_pct=Decimal("0.02"), bid_depth_notional_5=Decimal("50000"), ask_depth_notional_5=Decimal("50000"), buy_impact_1k_pct=Decimal("0.02"), sell_impact_1k_pct=Decimal("0.02")) original = build(c15) revised = [dict(row) for row in c15] revised[-2]["c"] = str(Decimal(str(revised[-2]["c"])) + Decimal("0.01")) changed = build(revised) assert changed.data_window_id != original.data_window_id def test_v77_long_and_short_retest_gates_are_directional(): long_feature = _feature() assert entry_blockers(long_feature, side="long") == [] short_feature = _feature( open_15m=Decimal("100.5"), close_15m=Decimal("100"), high_15m=Decimal("101.2"), low_15m=Decimal("99.5"), move_15m_pct=Decimal("-0.50"), move_1h_pct=Decimal("-0.60"), sma_1h_fast=Decimal("101"), sma_1h_slow=Decimal("103"), sma_4h_fast=Decimal("104"), sma_4h_slow=Decimal("106"), strength_rank=4, weakness_rank=1, breadth_positive_pct=Decimal("25"), recent_high=Decimal("101"), recent_low=Decimal("99"), pullback_from_high_pct=Decimal("0.99"), rebound_from_low_pct=Decimal("1.01"), rsi_15m=Decimal("40"), ) assert entry_blockers(short_feature, side="short") == [] assert "trend_1h_not_bearish" in entry_blockers(long_feature, side="short") def test_tradingview_bearish_alignment_becomes_short_paper_opportunity(): tv = TradingViewContext(coin="BTC", symbol="BTCUSDT", bias="bearish", confidence=Decimal("0.82"), trend_alignment="aligned", volume_state="breakout", volatility_state="expanding", source_age_seconds=30) impulse = classify_tradingview_impulse(tv, side="short") assert impulse["category"] == "sell_opportunity" assert impulse["action"] == "paper_candidate_if_other_gates_pass" assert impulse["live_order_allowed"] is False def test_partial_tp_moves_stop_to_break_even_and_keeps_half_open(tmp_path): state = {"open_positions": {"BTC": { "side": "long", "entry": "100", "size": "0.10", "original_size": "0.10", "stop_loss": "99", "tp1": "101.2", "tp1_hit": False, "entry_fee_usd": "0.0045", "spread_cost_usd": "0.001", "slippage_cost_usd": "0.001", "exit_slippage_pct": "0.02", "funding_rate_hourly_pct": "0", "last_managed_candle_ts": 100, "bars_held": 0, "mfe_pct": "0", "mae_pct": "0", "high_watermark": "100", "low_watermark": "100", "realized_net_pnl_accum": "0", "data_window_id": "w1", "setup": "confirmed_trend_retest_anti_chase_long", }}} feature = _feature(candle_ts=200, high_15m=Decimal("101.3"), low_15m=Decimal("99.5"), close_15m=Decimal("101")) closed, _ = runtime._manage_positions(state=state, features={"BTC": feature}, runtime_dir=tmp_path, strategy_id=LONG_STRATEGY_ID, run_id="run-1") assert closed == 0 pos = state["open_positions"]["BTC"] assert pos["tp1_hit"] is True assert Decimal(pos["size"]) == Decimal("0.050") assert Decimal(pos["stop_loss"]) == Decimal("100") rows = [json.loads(line) for line in (tmp_path / "trade_journal.jsonl").read_text().splitlines()] assert rows[-1]["event"] == "partial_exit" assert rows[-1]["exit_reason"] == "tp1_partial_50pct" def test_runtime_refuses_any_live_or_signed_flag(monkeypatch): monkeypatch.setenv("CTB_PAPER_TRADING", "true") monkeypatch.setenv("CTB_LIVE_TRADING_ALLOWED", "false") monkeypatch.setenv("HL_MAINNET_SIGNED_ACTION", "false") monkeypatch.setenv("CTB_LIVE_ORDER_ALLOWED", "true") with pytest.raises(PermissionError): runtime.scan_once(LONG_STRATEGY_ID, ["BTC"], md=object()) def test_scan_once_with_real_feature_shapes_stays_paper_only(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) monkeypatch.setenv("CTB_PAPER_TRADING", "true") monkeypatch.setenv("CTB_LIVE_TRADING_ALLOWED", "false") monkeypatch.setenv("CTB_LIVE_ORDER_ALLOWED", "false") monkeypatch.setenv("HL_MAINNET_SIGNED_ACTION", "false") now_ms = 2_000_000_000_000 class FakeMarketData: def get_meta_and_asset_ctxs(self): return [{"universe": [{"name": "BTC"}]}, [{"funding": "0.0001", "openInterest": "1000", "premium": "0"}]] def get_candles(self, coin, interval, start_ms, end_ms): if interval == "15m": return _candles(interval_ms=900_000, closes=[Decimal("99.5") for _ in range(59)] + [Decimal("100")], now_ms=now_ms, latest={"o": "99.4", "h": "100.4", "l": "98.8", "c": "100", "v": "10000"}) if interval == "1h": return _candles(interval_ms=3_600_000, closes=[Decimal("90") + Decimal(i) / Decimal("5") for i in range(60)], now_ms=now_ms) return _candles(interval_ms=14_400_000, closes=[Decimal("80") + Decimal(i) / Decimal("3") for i in range(60)], now_ms=now_ms) def get_l2_book(self, coin): return {"levels": [ [{"px": "99.99", "sz": "100"}, {"px": "99.98", "sz": "100"}, {"px": "99.97", "sz": "100"}, {"px": "99.96", "sz": "100"}, {"px": "99.95", "sz": "100"}], [{"px": "100.01", "sz": "100"}, {"px": "100.02", "sz": "100"}, {"px": "100.03", "sz": "100"}, {"px": "100.04", "sz": "100"}, {"px": "100.05", "sz": "100"}], ]} monkeypatch.setattr(runtime.time, "time", lambda: now_ms / 1000) payload = runtime.scan_once(LONG_STRATEGY_ID, ["BTC"], md=FakeMarketData()) assert payload["status"] == "ok" assert payload["features_loaded"] == 1 assert payload["paper_trading"] is True assert payload["live_order_allowed"] is False signal = json.loads((Path(payload["runtime_dir"]) / "signal_journal.jsonl").read_text().splitlines()[-1]) assert signal["proxy_inputs_used"] is False assert signal["features"]["data_source"] == "hyperliquid_live_readonly_ohlcv_l2" def test_runtime_requires_explicit_paper_flag(monkeypatch): monkeypatch.delenv("CTB_PAPER_TRADING", raising=False) with pytest.raises(PermissionError): runtime.scan_once(LONG_STRATEGY_ID, ["BTC"], md=object()) def test_runtime_routes_v77_2_momentum_as_separate_long_research_version(): feature = _feature( coin="ETH", current_price=Decimal("102"), close_15m=Decimal("102"), open_15m=Decimal("101.5"), recent_high=Decimal("101.8"), volume_ratio=Decimal("1.15"), move_1h_pct=Decimal("1.2"), move_15m_pct=Decimal("0.49"), rsi_15m=Decimal("64"), sma_1h_fast=Decimal("101"), sma_1h_slow=Decimal("99"), sma_4h_fast=Decimal("98"), sma_4h_slow=Decimal("96"), strength_rank=1, breadth_positive_pct=Decimal("75"), continuation_closes_confirmed=True, ) assert runtime.SUPPORTED[MOMENTUM_RESEARCH_ID] == "long" assert runtime._strategy_version(MOMENTUM_RESEARCH_ID) == MOMENTUM_VERSION assert runtime._strategy_entry_blockers(MOMENTUM_RESEARCH_ID, feature, tv=None) == [] intent = runtime._strategy_intent(MOMENTUM_RESEARCH_ID, feature, side="long", client_order_id="m") assert intent is not None assert intent.strategy_id == MOMENTUM_RESEARCH_ID def test_momentum_uses_universal_confluence_risk_not_generic_long_recommendation(tmp_path): report = tmp_path / "confluence.json" report.write_text(json.dumps({"confluence": {"scores": {"ETH": { "recommendation": "block_new_entry", "blockers": ["confluence_below_threshold"], "components": {"reliability": "0.95"}, }}}}), encoding="utf-8") assert runtime._confluence_blockers("ETH", "long", report, strategy_id=MOMENTUM_RESEARCH_ID) == [] report.write_text(json.dumps({"confluence": {"scores": {"ETH": { "recommendation": "block_new_entry", "blockers": ["event_risk_red"], "components": {"reliability": "0.95"}, }}}}), encoding="utf-8") assert runtime._confluence_blockers("ETH", "long", report, strategy_id=MOMENTUM_RESEARCH_ID) == ["confluence_event_risk_red"]