aaron/alphax
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix bug

Browse Source
This commit is contained in:
aaron 2026-05-14 11:41:41 +08:00
parent c7786946af
commit 7523f2a48f
2 changed files with 67 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
app/core/opportunity_lifecycle.py
View File

@ -316,6 +316,27 @@ def apply_entry_quality_gate(
if change_24h > _cfg_value(cfg, "gain_24h_wait_pct") and rr1 < 1.5:
reasons.append(f"24h涨幅{change_24h:.1f}%且rr1不足，禁止追涨")
if action_status == "等回踩" and current_price > 0:
plan_entry_price = to_float(entry_plan.get("entry_price"))
if plan_entry_price > 0:
wait_deviation_pct = round((current_price / plan_entry_price - 1) * 100, 2)
entry_plan["wait_pullback_deviation_pct"] = wait_deviation_pct
lifecycle_plan_type = ((entry_plan.get("opportunity_lifecycle") or {}).get("plan_type") or "").strip()
# 回踩参考已经被有效击穿，继续挂“等回踩”会误导；先降为观察。
if wait_deviation_pct < -1.2:
target_action = "观察"
reasons.append("回踩参考已下破，转为观察")
# 参考价已到或更优，且 RR 达标时，直接转为入场窗口。
elif wait_deviation_pct <= 0.3 and lifecycle_plan_type != "ambush":
if risk_reward_ok is False or rr1 < _cfg_value(cfg, "min_rr_buy_now"):
target_action = "观察"
reasons.append("回踩参考已到，但实时盈亏比不达标，转为观察")
else:
target_action = "可即刻买入"
entry_plan["entry_trigger_confirmed"] = True
entry_plan["entry_action"] = "可即刻买入"
reasons.append("回踩参考已到或更优，转为入场窗口")
if breakout_distance > _cfg_value(cfg, "breakout_distance_ban_pct"):
target_action = "观察"
reasons.append(f"离突破位+{breakout_distance:.1f}%>{ _cfg_value(cfg, 'breakout_distance_ban_pct') }%，严禁现价追")
@ -327,12 +348,16 @@ def apply_entry_quality_gate(
elif reasons:
# 只要买点质量闸门明确给出拦截原因，就不能继续保留“可即刻买入”。
# 如果当前已经回到/跌破计划参考价，但实时 RR 仍不足，说明不是“等回踩”，而是“回踩到了也不值得买”，应降级为观察。
if action_status == "等回踩" and current_price > 0 and to_float(entry_plan.get("entry_price")) > 0 and current_price <= to_float(entry_plan.get("entry_price")) * 1.003 and (risk_reward_ok is False or rr1 < _cfg_value(cfg, "min_rr_buy_now")):
if any("回踩参考已下破" in str(x) for x in reasons):
target_action = "观察"
elif any("回踩参考已到或更优" in str(x) for x in reasons):
target_action = "可即刻买入"
elif action_status == "等回踩" and current_price > 0 and to_float(entry_plan.get("entry_price")) > 0 and current_price <= to_float(entry_plan.get("entry_price")) * 1.003 and (risk_reward_ok is False or rr1 < _cfg_value(cfg, "min_rr_buy_now")):
target_action = "观察"
reasons.append("回踩参考已到，但实时盈亏比不达标，转为观察")
elif action_status == "可即刻买入" and current_price > 0 and stop_loss > 0 and tp1 > stop_loss and (risk_reward_ok is False or rr1 < _cfg_value(cfg, "min_rr_buy_now")):
rr_target_entry = calc_rr_target_entry(stop_loss, tp1, _cfg_value(cfg, "min_rr_buy_now"))
if rr_target_entry > stop_loss and rr_target_entry < current_price * 0.997:
if rr_target_entry > stop_loss and rr_target_entry < current_price:
target_action = "等回踩"
entry_plan["entry_price"] = rr_target_entry
entry_plan["entry_method"] = f"等回踩至可买RR价 {rr_target_entry:.8g}"

40
tests/test_recommendation_execution_status.py
View File

@ -114,6 +114,46 @@ class RecommendationExecutionStatusTests(unittest.TestCase):
self.assertEqual(target['execution_label'], '🟡 等回踩，不追高')
self.assertIn('等待回踩', target['execution_reason'])
def test_wait_pullback_downbreak_degrades_to_observe(self):
self._insert_rec(
symbol='BBB/USDT',
action_status='等回踩',
current_price=98.0,
entry_plan_json=json.dumps({
'entry_price': 100.0,
'entry_action': '等回踩',
'risk_reward_ok': True,
'stop_loss': 95.0,
'tp1': 110.0,
'rr1': 2.0,
}, ensure_ascii=False),
)
rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False)
target = next(r for r in rows if r['symbol'] == 'BBB/USDT')
self.assertEqual(target['action_status'], '观察')
self.assertEqual(target['execution_status'], 'observe')
self.assertIn('回踩参考已下破', target['execution_reason'])
def test_wait_pullback_at_reference_promotes_to_buy_now_when_rr_ok(self):
self._insert_rec(
symbol='BBB/USDT',
action_status='等回踩',
current_price=99.9,
entry_plan_json=json.dumps({
'entry_price': 100.0,
'entry_action': '等回踩',
'risk_reward_ok': True,
'stop_loss': 95.0,
'tp1': 110.0,
'rr1': 2.0,
}, ensure_ascii=False),
)
rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False)
target = next(r for r in rows if r['symbol'] == 'BBB/USDT')
self.assertEqual(target['action_status'], '可即刻买入')
self.assertEqual(target['execution_status'], 'buy_now')
self.assertTrue(target['entry_plan']['entry_trigger_confirmed'])
def test_invalid_status_for_decay(self):
self._insert_rec(
symbol='CCC/USDT',