117 lines
5.5 KiB
Python
117 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.quant_engine.exit_decisions import compute_sell_decision
|
|
from src.quant_engine.exit_decisions import compute_stop_action_ladder
|
|
from src.quant_engine.exit_decisions import normalize_tick
|
|
|
|
|
|
class TestPriceQtyParityV1(unittest.TestCase):
|
|
def test_tp1_price_and_qty_parity(self):
|
|
res = compute_sell_decision({"close": 10000, "profitPct": 10.0, "tp1Price": 11000})
|
|
self.assertEqual(res["action"], "TAKE_PROFIT_TIER1")
|
|
self.assertEqual(res["ratio_pct"], 25)
|
|
self.assertEqual(res["price_basis"], "TAKE_PROFIT_TIER1_PRICE")
|
|
self.assertEqual(res["limit_price"], 11000)
|
|
self.assertEqual(res["order_type"], "LIMIT_SELL")
|
|
|
|
def test_tp2_price_and_fallback_parity(self):
|
|
res_fallback = compute_sell_decision({"close": 10000, "profitPct": 50.0, "tp2Price": None})
|
|
self.assertEqual(res_fallback["price_basis"], "PRIOR_CLOSE_X_0.998")
|
|
self.assertEqual(res_fallback["action"], "PROFIT_TRIM_50")
|
|
|
|
res_tp2 = compute_sell_decision({"close": 10000, "profitPct": 50.0, "tp2Price": 12000})
|
|
self.assertEqual(res_tp2["price_basis"], "TAKE_PROFIT_TIER2_PRICE")
|
|
self.assertEqual(res_tp2["limit_price"], 12000)
|
|
|
|
def test_relative_weakness_and_time_exit_price_qty_parity(self):
|
|
res_rw = compute_sell_decision({"close": 10000, "rwPartial": 1})
|
|
self.assertEqual(res_rw["action"], "TRIM_25")
|
|
self.assertEqual(res_rw["ratio_pct"], 25)
|
|
self.assertEqual(res_rw["price_basis"], "PRIOR_CLOSE_X_0.998")
|
|
|
|
res_rw2 = compute_sell_decision({"close": 10000, "rwPartial": 2})
|
|
self.assertEqual(res_rw2["action"], "TRIM_50")
|
|
self.assertEqual(res_rw2["ratio_pct"], 50)
|
|
self.assertEqual(res_rw2["order_type"], "LIMIT_SELL")
|
|
|
|
res_time = compute_sell_decision({"close": 10000, "daysToTimeStop": 0})
|
|
self.assertEqual(res_time["action"], "TIME_EXIT_100")
|
|
self.assertEqual(res_time["ratio_pct"], 100)
|
|
self.assertEqual(res_time["price_basis"], "TIME_STOP_CLOSE_PROTECT")
|
|
|
|
res_time_trim = compute_sell_decision({"close": 10000, "daysToTimeStop": 6})
|
|
self.assertEqual(res_time_trim["action"], "TIME_TRIM_50")
|
|
self.assertEqual(res_time_trim["ratio_pct"], 50)
|
|
|
|
res_time_window = compute_sell_decision({"close": 10000, "daysToTimeStop": 7})
|
|
self.assertEqual(res_time_window["action"], "TIME_TRIM_50")
|
|
self.assertEqual(res_time_window["ratio_pct"], 50)
|
|
|
|
res_time_14 = compute_sell_decision({"close": 10000, "daysToTimeStop": 14})
|
|
self.assertEqual(res_time_14["action"], "TIME_TRIM_25")
|
|
self.assertEqual(res_time_14["ratio_pct"], 25)
|
|
|
|
res_time_15 = compute_sell_decision({"close": 10000, "daysToTimeStop": 15})
|
|
self.assertEqual(res_time_15["action"], "HOLD")
|
|
|
|
def test_stop_action_ladder_parity(self):
|
|
res = compute_stop_action_ladder({"profitPct": 10.0})
|
|
self.assertEqual(res["action"], "TAKE_PROFIT_TIER1")
|
|
self.assertEqual(res["quantity_pct"], 25)
|
|
self.assertEqual(res["priority"], 5)
|
|
|
|
res_review = compute_stop_action_ladder({"profitPct": 9.99, "daysToTimeStop": 1})
|
|
self.assertEqual(res_review["action"], "REVIEW_HUMAN")
|
|
self.assertEqual(res_review["quantity_pct"], 0)
|
|
|
|
res_exit = compute_stop_action_ladder({"timingAction": "STOP_OR_TIME_EXIT_READY"})
|
|
self.assertEqual(res_exit["action"], "EXIT_100")
|
|
self.assertEqual(res_exit["quantity_pct"], 100)
|
|
|
|
res_risk_off = compute_stop_action_ladder({"REGIME_PRELIM": "RISK_OFF"})
|
|
self.assertEqual(res_risk_off["action"], "REGIME_TRIM_50")
|
|
self.assertEqual(res_risk_off["quantity_pct"], 50)
|
|
self.assertEqual(res_risk_off["priority"], 2)
|
|
|
|
res_rw2b = compute_stop_action_ladder({"rw_partial_excluding_rw2b": 1, "RW2b_5d_rapid_weakness": True})
|
|
self.assertEqual(res_rw2b["action"], "TRIM_50")
|
|
self.assertEqual(res_rw2b["priority"], 2.5)
|
|
|
|
res_trailing = compute_stop_action_ladder({"trailingStopBreach": True})
|
|
self.assertEqual(res_trailing["action"], "TRIM_50")
|
|
self.assertEqual(res_trailing["priority"], 4)
|
|
|
|
def test_fallback_stop_price_is_tick_independent(self):
|
|
res = compute_sell_decision({"close": 10000, "profitPct": 20.0, "tp1Price": None})
|
|
self.assertEqual(res["action"], "PROFIT_TRIM_25")
|
|
self.assertEqual(res["price_basis"], "PRIOR_CLOSE_X_0.998")
|
|
self.assertEqual(res["validation"], "SIGNAL_CONFIRMED")
|
|
|
|
res_tp2 = compute_sell_decision({"close": 10000, "profitPct": 50.0, "tp2Price": None})
|
|
self.assertEqual(res_tp2["action"], "PROFIT_TRIM_50")
|
|
self.assertEqual(res_tp2["price_basis"], "PRIOR_CLOSE_X_0.998")
|
|
|
|
res_tp2_fallback = compute_sell_decision({"close": 10000, "profitPct": 50.0, "tp2Price": None, "tp1Price": None})
|
|
self.assertEqual(res_tp2_fallback["action"], "PROFIT_TRIM_50")
|
|
self.assertEqual(res_tp2_fallback["price_source"], "CLOSE_PROFIT_PROTECT")
|
|
|
|
def test_tick_normalization_boundaries(self):
|
|
self.assertEqual(normalize_tick(1999.9), 1999)
|
|
self.assertEqual(normalize_tick(2000.0), 2000)
|
|
self.assertEqual(normalize_tick(5001.0), 5000)
|
|
self.assertEqual(normalize_tick(20000.0), 20000)
|
|
self.assertEqual(normalize_tick(50000.0), 50000)
|
|
self.assertEqual(normalize_tick(200000.0), 200000)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|