yuqei/ai-exchange
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 4 management rule behavior

Browse Source
This commit is contained in:
Codex
2026-05-06 15:13:53 +08:00
parent 3b2d1636e2
commit 79e21aeee4
4 changed files with 162 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/backtest/backtest_service.py
+70 -5
View File
@@ -385,15 +385,20 @@ def _evaluate_signal_outcome_steps(
replay_config: Optional[BacktestReplayConfig],
) -> BacktestOutcome:
entry_mid = (signal.entry_low + signal.entry_high) / 2
management_rules = list(signal.management_rules or [])
tp1_partial_fraction = resolve_tp1_partial_fraction(management_rules)
move_stop_enabled = has_management_rule(management_rules, "move_stop")
entry_triggered = False
tp1_hit = False
tp2_hit = False
stopped_out = False
stopped_after_tp1 = False
same_candle_ambiguous = False
mfe = 0.0
mae = 0.0
entry_step: Optional[_BacktestEvaluationStep] = None
exit_step: Optional[_BacktestEvaluationStep] = None
active_stop_loss = signal.stop_loss
for step in steps:
candle = step.candle
@@ -413,13 +418,13 @@ def _evaluate_signal_outcome_steps(
if signal.side == 'buy':
mfe = max(mfe, high - entry_mid)
mae = max(mae, entry_mid - low)
stop_hit = bool(signal.stop_loss and low <= signal.stop_loss)
stop_hit = bool(active_stop_loss and low <= active_stop_loss)
tp1_now = bool(signal.tp1 and high >= signal.tp1)
tp2_now = bool(signal.tp2 and high >= signal.tp2)
else:
mfe = max(mfe, entry_mid - low)
mae = max(mae, high - entry_mid)
stop_hit = bool(signal.stop_loss and high >= signal.stop_loss)
stop_hit = bool(active_stop_loss and high >= active_stop_loss)
tp1_now = bool(signal.tp1 and low <= signal.tp1)
tp2_now = bool(signal.tp2 and low <= signal.tp2)
@@ -430,10 +435,13 @@ def _evaluate_signal_outcome_steps(
if stop_hit:
stopped_out = True
stopped_after_tp1 = tp1_hit
exit_step = step
break
if tp1_now:
tp1_hit = True
if move_stop_enabled:
active_stop_loss = entry_mid
if tp2_now:
tp2_hit = True
exit_step = step
@@ -448,6 +456,8 @@ def _evaluate_signal_outcome_steps(
error_tags.append('entry_not_triggered')
elif tp2_hit:
outcome = 'win'
elif stopped_out and tp1_hit and move_stop_enabled:
outcome = 'breakeven'
elif tp1_hit and not stopped_out:
outcome = 'breakeven'
elif stopped_out:
@@ -471,7 +481,23 @@ def _evaluate_signal_outcome_steps(
rr_tp1=signal.rr_tp1,
rr_tp2=signal.rr_tp2,
cost_r=cost_r,
tp1_hit=tp1_hit,
tp2_hit=tp2_hit,
tp1_partial_fraction=tp1_partial_fraction,
stopped_after_tp1=stopped_after_tp1,
move_stop_applied=move_stop_enabled,
)
remaining_position_fraction = max(0.0, 1.0 - tp1_partial_fraction) if tp1_hit else 1.0
management_applied = {
"tp1_partial": tp1_partial_fraction > 0 and tp1_hit,
"tp1_partial_fraction": tp1_partial_fraction if tp1_hit else 0.0,
"move_stop": move_stop_enabled and tp1_hit,
"move_stop_to": "breakeven" if move_stop_enabled and tp1_hit else None,
"remaining_position_fraction": remaining_position_fraction,
"remaining_position_closed": bool(tp2_hit or (stopped_after_tp1 and move_stop_enabled)),
"tp1_realized_r": (tp1_partial_fraction * signal.rr_tp1) if tp1_hit else 0.0,
"active_stop_loss_after_tp1": entry_mid if move_stop_enabled and tp1_hit else signal.stop_loss,
}
meta = {
'entry_triggered': entry_triggered,
@@ -488,6 +514,7 @@ def _evaluate_signal_outcome_steps(
'cost_model': cost_config.__dict__,
'backtest_mode': mode,
'r_multiple': r_multiple,
'management_applied': management_applied,
}
if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD:
meta.update(
@@ -570,12 +597,34 @@ def estimate_exit_price(signal: BacktestSignal, outcome: str, entry_price: float
return entry_price
def compute_outcome_r_multiple(outcome: str, rr_tp1: float = 0.0, rr_tp2: float = 0.0, cost_r: float = 0.0) -> float:
def compute_outcome_r_multiple(
outcome: str,
rr_tp1: float = 0.0,
rr_tp2: float = 0.0,
cost_r: float = 0.0,
tp1_hit: bool = False,
tp2_hit: bool = False,
tp1_partial_fraction: float = 0.0,
stopped_after_tp1: bool = False,
move_stop_applied: bool = False,
) -> float:
"""Convert the current simplified outcome labels into conservative R multiples."""
if outcome == 'win':
gross_r = rr_tp2
if tp1_hit and tp1_partial_fraction > 0:
gross_r = (tp1_partial_fraction * rr_tp1) + ((1.0 - tp1_partial_fraction) * rr_tp2)
else:
gross_r = rr_tp2
elif outcome == 'loss':
gross_r = -1.0
if tp1_hit and tp1_partial_fraction > 0:
remaining_loss_r = 0.0 if move_stop_applied and stopped_after_tp1 else -(1.0 - tp1_partial_fraction)
gross_r = (tp1_partial_fraction * rr_tp1) + remaining_loss_r
else:
gross_r = -1.0
elif outcome == 'breakeven':
if tp1_hit and tp1_partial_fraction > 0:
gross_r = tp1_partial_fraction * rr_tp1
else:
gross_r = 0.0
else:
gross_r = 0.0
@@ -584,6 +633,22 @@ def compute_outcome_r_multiple(outcome: str, rr_tp1: float = 0.0, rr_tp2: float
return gross_r
def has_management_rule(management_rules: list[dict], code: str) -> bool:
return any(str(item.get("code") or "") == code for item in management_rules)
def resolve_tp1_partial_fraction(management_rules: list[dict]) -> float:
for item in management_rules:
if str(item.get("code") or "") != "tp1_partial":
continue
try:
fraction = float(item.get("fraction") or 0.0)
except (TypeError, ValueError):
return 0.0
return min(max(fraction, 0.0), 1.0)
return 0.0
def classify_backtest_lifecycle_state(
*,
outcome: Optional[str],
src/services/execution/execution_service.py
+16 -1
View File
@@ -489,6 +489,18 @@ def evaluate_account_risk_limits(account_state: Optional[AccountRiskState]) -> l
def build_execution_observation_from_backtest_result(backtest_result: Optional[dict]) -> dict:
backtest_lifecycle_state = classify_backtest_lifecycle_state_from_result(backtest_result)
meta = dict((backtest_result or {}).get("meta") or {})
management_applied = dict(meta.get("management_applied") or {})
remaining_closed = bool(management_applied.get("remaining_position_closed"))
close_reason = backtest_lifecycle_state
status = None
lifecycle_state = backtest_lifecycle_state
position_state = None
if management_applied.get("tp1_partial") and remaining_closed:
status = "closed"
lifecycle_state = "tp1_partial_managed_to_close"
position_state = "closed"
close_reason = "tp1_partial_then_management_close"
return {
"source": "backtest_projection",
"entry_triggered": bool(backtest_result and (backtest_result.get("outcome") in {"win", "loss", "breakeven"} or backtest_result.get("tp1_hit") or backtest_result.get("tp2_hit"))),
@@ -497,7 +509,10 @@ def build_execution_observation_from_backtest_result(backtest_result: Optional[d
"stop_loss_hit": backtest_lifecycle_state == BACKTEST_LIFECYCLE_STOPPED_OUT,
"invalidated_before_entry": bool(backtest_result and backtest_result.get("invalidated_before_entry")),
"backtest_lifecycle_state": backtest_lifecycle_state,
"close_reason": backtest_lifecycle_state,
"close_reason": close_reason,
"status": status,
"lifecycle_state": lifecycle_state,
"position_state": position_state,
}
tests/unit/test_backtest_service.py
+45
View File
@@ -124,6 +124,18 @@ class BacktestOutcomeTests(unittest.TestCase):
self.assertAlmostEqual(compute_outcome_r_multiple("win", rr_tp2=2.5, cost_r=0.1), 2.4)
self.assertAlmostEqual(compute_outcome_r_multiple("loss", cost_r=0.1), -1.1)
self.assertEqual(compute_outcome_r_multiple("missed", cost_r=0.1), 0.0)
self.assertAlmostEqual(
compute_outcome_r_multiple(
"breakeven",
rr_tp1=1.0,
cost_r=0.1,
tp1_hit=True,
tp1_partial_fraction=0.5,
stopped_after_tp1=True,
move_stop_applied=True,
),
0.4,
)
def test_cost_model_reduces_executed_r_multiple(self) -> None:
signal = BacktestSignal(
@@ -160,6 +172,39 @@ class BacktestOutcomeTests(unittest.TestCase):
self.assertAlmostEqual(outcome.meta["r_multiple"], 2.5 - expected_cost_r)
self.assertEqual(outcome.meta["cost_model"]["fee_rate"], 0.0005)
def test_tp1_partial_and_move_stop_affect_backtest_r_multiple(self) -> None:
signal = BacktestSignal(
signal_id=40,
side="buy",
entry_low=100.0,
entry_high=101.0,
stop_loss=99.0,
tp1=102.0,
tp2=104.0,
rr_tp1=1.0,
rr_tp2=2.3333333333,
management_rules=[
{"code": "tp1_partial", "fraction": 0.5},
{"code": "move_stop", "to": "breakeven"},
],
)
outcome = evaluate_signal_outcome(
signal,
[
{"high": 102.1, "low": 100.2},
{"high": 101.0, "low": 100.4},
],
)
self.assertEqual(outcome.outcome, "breakeven")
self.assertTrue(outcome.tp1_hit)
self.assertFalse(outcome.tp2_hit)
self.assertAlmostEqual(outcome.meta["r_multiple"], 0.5)
self.assertTrue(outcome.meta["management_applied"]["tp1_partial"])
self.assertTrue(outcome.meta["management_applied"]["move_stop"])
self.assertTrue(outcome.meta["management_applied"]["remaining_position_closed"])
def test_replay_no_lookahead_tracks_visible_candles_and_confirmed_swings_at_entry(self) -> None:
signal = BacktestSignal(
signal_id=5,
tests/unit/test_execution_service.py
+31
View File
@@ -338,6 +338,22 @@ class ExecutionAuditTests(unittest.TestCase):
self.assertFalse(settled["meta"]["stop_loss_hit"])
self.assertEqual(settled["meta"]["position_state"], "partially_closed")
def test_settle_execution_from_backtest_closes_when_management_finishes_remaining_position(self) -> None:
settled = ExecutionService().settle_execution_from_backtest(
trade_execution={"status": "open", "meta": {}, "closed_ts": None},
backtest_result={
"outcome": "breakeven",
"tp1_hit": True,
"tp2_hit": False,
"invalidated_before_entry": False,
"meta": {"management_applied": {"tp1_partial": True, "remaining_position_closed": True}},
},
)
self.assertEqual(settled["status"], "closed")
self.assertEqual(settled["meta"]["lifecycle_state"], "tp1_partial_managed_to_close")
self.assertEqual(settled["meta"]["position_state"], "closed")
def test_settle_execution_from_backtest_closes_invalidated_before_entry(self) -> None:
settled = ExecutionService().settle_execution_from_backtest(
trade_execution={"status": "open", "meta": {}, "closed_ts": None},
@@ -369,6 +385,21 @@ class ExecutionAuditTests(unittest.TestCase):
self.assertEqual(observation["close_reason"], "tp2_closed")
self.assertTrue(observation["entry_triggered"])
def test_build_execution_observation_closes_partial_trade_when_management_finishes_remaining_position(self) -> None:
observation = build_execution_observation_from_backtest_result(
{
"outcome": "breakeven",
"tp1_hit": True,
"tp2_hit": False,
"invalidated_before_entry": False,
"meta": {"management_applied": {"tp1_partial": True, "remaining_position_closed": True}},
}
)
self.assertEqual(observation["status"], "closed")
self.assertEqual(observation["lifecycle_state"], "tp1_partial_managed_to_close")
self.assertEqual(observation["position_state"], "closed")
def test_persist_execution_settlement_updates_repository_when_execution_has_id(self) -> None:
repository = Mock()
service = ExecutionService(repository=repository)