diff --git a/backend/src/alphapulse/trading/analyst.py b/backend/src/alphapulse/trading/analyst.py index eabcb0a..6936d1b 100644 --- a/backend/src/alphapulse/trading/analyst.py +++ b/backend/src/alphapulse/trading/analyst.py @@ -285,6 +285,12 @@ async def submit(prepared: Prepared, symbol: str, analysis_id: int, arguments: d snapshot = await read_snapshot() if snapshot: prepared.context.position_side = position_side(snapshot, symbol) + prepared.context.closed_prices = {} + for interval in sorted({c.timeframe for c in plan.reevaluate if c.timeframe != "mark"}): + rows = await market_client.get("um", "klines", {"symbol": symbol, "interval": interval, "limit": 2}) + now_ms = datetime.now(UTC).timestamp() * 1000 + closed = rows[-1] if rows[-1][6] < now_ms else rows[-2] + prepared.context.closed_prices[interval] = Decimal(closed[4]) try: sizing = check_setup(plan, prepared.context) except PlanError as e: diff --git a/backend/src/alphapulse/trading/plan.py b/backend/src/alphapulse/trading/plan.py index 7759ae0..8c0feb0 100644 --- a/backend/src/alphapulse/trading/plan.py +++ b/backend/src/alphapulse/trading/plan.py @@ -5,7 +5,7 @@ 下方常量与 main/README 的数字重复，并写进了分析器的提示词；改 README 时要同步修改。 """ -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal from typing import Annotated, Literal @@ -98,6 +98,8 @@ class PlanContext: atr: Decimal has_plan: bool position_side: Direction | None + # 非 mark 复核条件使用的最近已收盘价，由提交边界按需获取。 + closed_prices: dict[str, Decimal] = field(default_factory=dict) @dataclass(frozen=True) @@ -204,9 +206,9 @@ def check_action(plan: TradePlan, ctx: PlanContext) -> list[str]: if plan.action == "plan" and ctx.position_side is not None: errors.append("已有持仓：持仓由程序按止损、止盈和持仓时限管理，只能 keep 或 exit") for i, c in enumerate(plan.reevaluate, 1): - satisfied = ctx.mark_price > c.price if c.direction == "above" else ctx.mark_price < c.price - if satisfied: - errors.append(f"reevaluate 第 {i} 个条件（{c.direction} {c.price}）按现价 {ctx.mark_price} 已经满足，提交后会立即触发") + price = ctx.mark_price if c.timeframe == "mark" else ctx.closed_prices[c.timeframe] + if condition_met(c.direction, c.price, price): + errors.append(f"reevaluate 第 {i} 个条件（{c.timeframe} {c.direction} {c.price}）按价格 {price} 已经满足，提交后会立即触发") return errors diff --git a/backend/tests/test_plan.py b/backend/tests/test_plan.py index 9ea75d8..5c62071 100644 --- a/backend/tests/test_plan.py +++ b/backend/tests/test_plan.py @@ -151,10 +151,26 @@ def test_action_rules(): assert check_setup(TradePlan(action="wait", summary="s", reasons="r", risks="x"), context()) == {} satisfied = TradePlan(action="wait", summary="s", reasons="r", risks="x", reevaluate=[{"direction": "above", "price": "80000", "timeframe": "5m", "basis": "x"}]) - assert "已经满足" in errors_of(satisfied, context()) + assert "已经满足" in errors_of(satisfied, context(closed_prices={"5m": Decimal("81000")})) assert "只能 keep 或 exit" in errors_of(setup_plan(), context(position_side="long", has_plan=True)) +@pytest.mark.parametrize("direction,mark,closed,satisfied", [ + ("above", "110", "100", False), ("above", "100", "110", True), + ("below", "100", "110", False), ("below", "110", "100", True), +]) +@pytest.mark.parametrize("timeframe", ["mark", "5m"]) +def test_reevaluate_uses_its_own_price(direction, mark, closed, satisfied, timeframe): + plan = TradePlan(action="wait", summary="s", reasons="r", risks="x", + reevaluate=[{"direction": direction, "price": "105", "timeframe": timeframe, "basis": "x"}]) + ctx = context(mark_price=Decimal(mark), closed_prices={"5m": Decimal(closed)}) + expected = satisfied if timeframe == "5m" else not satisfied + if expected: + assert "已经满足" in errors_of(plan, ctx) + else: + assert check_setup(plan, ctx) == {} + + # ---------- 应有订单 ---------- LIMIT_PLAN = {"direction": "long", "setup": {}, "entry": {"price": "80500.0", "quantity": "0.0241", "stop": "80150.0"}} diff --git a/backend/tests/test_trading_safety.py b/backend/tests/test_trading_safety.py index a65ef7b..77a8c5b 100644 --- a/backend/tests/test_trading_safety.py +++ b/backend/tests/test_trading_safety.py @@ -2,6 +2,7 @@ import asyncio from datetime import UTC, datetime, timedelta from decimal import Decimal from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest @@ -288,3 +289,33 @@ def test_replacement_proposal_survives_old_plan_cancellation(trading): finally: await engine.market.http.aclose() asyncio.run(check()) + + +@pytest.mark.parametrize("closed,mark,rejected", [("100", "110", False), ("110", "100", True)]) +@pytest.mark.parametrize("last_bar_closed", [False, True]) +def test_submit_reevaluate_uses_latest_closed_candle(trading, monkeypatch, closed, mark, rejected, last_bar_closed): + from alphapulse.trading import analyst + from alphapulse.trading.plan import PlanContext + + now_ms = int(datetime.now(UTC).timestamp() * 1000) + rows = [[0, "0", "0", "0", closed, "0", now_ms - 1000], + [1, "0", "0", "0", mark, "0", now_ms + 60_000]] + if last_bar_closed: + rows[-1][4], rows[-1][6] = closed, now_ms - 1 + monkeypatch.setattr(analyst.market_client, "get", AsyncMock(return_value=rows)) + monkeypatch.setattr(analyst, "mark_price", AsyncMock(return_value=Decimal(mark))) + monkeypatch.setattr(analyst, "recent_atr", AsyncMock(return_value=Decimal(1))) + monkeypatch.setattr(analyst, "read_snapshot", AsyncMock(return_value=None)) + monkeypatch.setattr(analyst, "add_event", AsyncMock()) + ctx = PlanContext(Decimal(mark), Decimal(1000), RULES, Decimal(1), False, None) + prepared = SimpleNamespace(exchange="bybit", context=ctx) + + async def check(): + result, outcome = await analyst.submit(prepared, "BTCUSDT", 1, { + "action": "wait", "summary": "s", "reasons": "r", "risks": "x", + "reevaluate": [{"direction": "above", "price": "105", "timeframe": "5m", "basis": "收盘确认"}], + }) + assert ctx.closed_prices == {"5m": Decimal(closed)} + assert (outcome is None) == rejected + assert ("已经满足" in result) == rejected + asyncio.run(check())