diff --git a/backend/src/alphapulse/api/trading.py b/backend/src/alphapulse/api/trading.py index e307936..18d0d01 100644 --- a/backend/src/alphapulse/api/trading.py +++ b/backend/src/alphapulse/api/trading.py @@ -17,7 +17,7 @@ from alphapulse.api.market import call_binance, client as market_client from alphapulse.core.auth import require_admin from alphapulse.core.db import get_session from alphapulse.core.redis import KEY_PREFIX, redis -from alphapulse.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingSymbol +from alphapulse.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol from alphapulse.trading.client import OWN_CLIENT_ID, ExchangeError, public_get from alphapulse.trading.plan import DAILY_ANALYSIS_LIMIT from alphapulse.trading.snapshot import read_snapshot @@ -30,6 +30,8 @@ SYMBOLS_CACHE_KEY = f"{KEY_PREFIX}trading:symbols" SYMBOLS_CACHE_SECONDS = 3600 # 删除或切换角度前要求快照足够新，否则无法确认没有仓位和挂单 SNAPSHOT_FRESH_SECONDS = 30 +# 执行器正在管理的计划状态 +LIVE_PLAN_STATUSES = ("pending", "open", "closing") class TodayUsage(BaseModel): @@ -70,8 +72,10 @@ class SymbolOut(BaseModel): added_at: datetime analysis_requested_at: datetime | None analyzing: bool - # 当前计划的状态：pending / open，没有计划为空 + # 当前计划：pending / open / closing，没有计划为空 + plan_id: int | None plan_status: str | None + close_deadline: datetime | None class SymbolIn(BaseModel): @@ -113,6 +117,28 @@ class AnalysisDetail(AnalysisOut): transcript: list | None +class OrderOut(BaseModel): + id: int + purpose: str + leg: int | None + side: str + type: str + price: Decimal | None + trigger_price: Decimal | None + quantity: Decimal + status: str + filled_qty: Decimal + avg_price: Decimal | None + error: str | None + created_at: datetime + + +class SymbolPlanOut(BaseModel): + # 执行器正在管理的计划；没有时为最近一次待采纳的提议 + plan: AnalysisOut | None + orders: list[OrderOut] + + class EventOut(BaseModel): id: int created_at: datetime @@ -236,14 +262,15 @@ async def list_symbol_options() -> list[SymbolOption]: async def list_symbols(session: Annotated[AsyncSession, Depends(get_session)]) -> list[SymbolOut]: symbols = (await session.scalars(select(TradingSymbol).order_by(TradingSymbol.added_at))).all() analyzing = set((await session.scalars(select(TradingAnalysis.symbol).where(TradingAnalysis.status == "running"))).all()) - plans = dict((await session.execute( - select(TradingAnalysis.symbol, TradingAnalysis.plan_status) - .where(TradingAnalysis.plan_status.in_(("pending", "open")), TradingAnalysis.dry_run.is_(False)) - )).all()) + plans = {p.symbol: p for p in await session.scalars( + select(TradingAnalysis).where(TradingAnalysis.plan_status.in_(LIVE_PLAN_STATUSES)).order_by(TradingAnalysis.id) + )} return [ SymbolOut( symbol=s.symbol, angle=s.angle, added_at=s.added_at, analysis_requested_at=s.analysis_requested_at, - analyzing=s.symbol in analyzing, plan_status=plans.get(s.symbol), + analyzing=s.symbol in analyzing, plan_id=plans[s.symbol].id if s.symbol in plans else None, + plan_status=plans[s.symbol].plan_status if s.symbol in plans else None, + close_deadline=plans[s.symbol].close_deadline if s.symbol in plans else None, ) for s in symbols ] @@ -269,8 +296,7 @@ async def ensure_idle(session: AsyncSession, symbol: str, action: str) -> Tradin raise HTTPException(status.HTTP_404_NOT_FOUND, f"{symbol} 不在交易列表中") busy = await session.scalar(select(func.count()).where( TradingAnalysis.symbol == symbol, - or_(TradingAnalysis.status == "running", - TradingAnalysis.plan_status.in_(("pending", "open")) & TradingAnalysis.dry_run.is_(False)), + or_(TradingAnalysis.status == "running", TradingAnalysis.plan_status.in_(("proposed", *LIVE_PLAN_STATUSES))), )) if busy: raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 有进行中的分析或计划，不能{action}") @@ -354,3 +380,18 @@ async def get_analysis(analysis_id: int, session: Annotated[AsyncSession, Depend if analysis is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "分析不存在") return AnalysisDetail.model_validate(analysis, from_attributes=True) + + +@router.get("/symbols/{symbol}/plan") +async def get_symbol_plan(symbol: str, session: Annotated[AsyncSession, Depends(get_session)]) -> SymbolPlanOut: + plan = await session.scalar( + select(TradingAnalysis).where(TradingAnalysis.symbol == symbol, TradingAnalysis.plan_status.in_(("proposed", *LIVE_PLAN_STATUSES))) + .order_by(TradingAnalysis.id.desc()).limit(1) + ) + if plan is None: + return SymbolPlanOut(plan=None, orders=[]) + orders = await session.scalars(select(TradingOrder).where(TradingOrder.analysis_id == plan.id).order_by(TradingOrder.id)) + return SymbolPlanOut( + plan=AnalysisOut.model_validate(plan, from_attributes=True), + orders=[OrderOut.model_validate(o, from_attributes=True) for o in orders], + ) diff --git a/backend/src/alphapulse/models/trading.py b/backend/src/alphapulse/models/trading.py index 72291e3..43ddd6e 100644 --- a/backend/src/alphapulse/models/trading.py +++ b/backend/src/alphapulse/models/trading.py @@ -69,7 +69,8 @@ class TradingAnalysis(Base): error: Mapped[str | None] = mapped_column(Text) started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) - # pending（挂单中）/ open（持仓中）/ closed / canceled（未成交即结束）/ replaced（被新计划替换） + # proposed（待执行器采纳）/ pending（挂单中）/ open（持仓中）/ closing（平仓中）/ closed / canceled（未成交即结束） + # / replaced（被新计划替换）/ stale（采纳前持仓已变，作废）/ applied（exit 决定已执行） plan_status: Mapped[str | None] = mapped_column(String(16)) valid_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) first_fill_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) diff --git a/backend/src/alphapulse/trading/analyst.py b/backend/src/alphapulse/trading/analyst.py index c1840ea..132bde6 100644 --- a/backend/src/alphapulse/trading/analyst.py +++ b/backend/src/alphapulse/trading/analyst.py @@ -6,7 +6,6 @@ import asyncio import json import logging -import time from dataclasses import dataclass from datetime import UTC, datetime from decimal import Decimal @@ -18,10 +17,10 @@ from alphapulse.ai import llm from alphapulse.ai.prompt import data_overview, estimate_tokens, skills_section from alphapulse.ai.tools import TOOLS, dump, local, run_tool from alphapulse.core.db import SessionLocal -from alphapulse.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingSymbol +from alphapulse.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol from alphapulse.trading import plan as rules -from alphapulse.trading.client import public_get -from alphapulse.trading.plan import PlanContext, PlanError, SymbolRules, TradePlan, size_plan +from alphapulse.trading.client import OWN_CLIENT_ID, ExchangeError, mark_price, symbol_rules +from alphapulse.trading.plan import PlanContext, PlanError, TradePlan, size_plan from alphapulse.trading.snapshot import read_snapshot logger = logging.getLogger("alphapulse.trading.analyst") @@ -30,7 +29,6 @@ READ_TOOLS = ("search_news", "get_news", "search_symbols", "get_klines", "get_ti # 最后一轮只开放 submit_trade_plan，逼模型给出结论 MAX_ROUNDS = 6 RECENT_DECISIONS = 2 -EXCHANGE_INFO_SECONDS = 3600 ANGLE_LABELS = {"short": "短线（24 小时内平仓）", "long": "长线（12–48 小时内平仓）"} DIRECTION_LABELS = {"long": "做多", "short": "做空"} CONFIDENCE_LABELS = {"high": "高", "medium": "中", "low": "低"} @@ -74,6 +72,11 @@ INSTRUCTIONS = f"""你是 AlphaPulse 的模拟盘交易分析器：为一个币 - 每档入场的名义价值不低于交易所最小下单金额，不足时合并或删档。 - 单笔风险为交易资金的 {_pct(rules.RISK_PCT['short'])}（短线）或 {_pct(rules.RISK_PCT['long'])}（长线），置信度低时减半；名义仓位合计不超过 {rules.MAX_LEVERAGE} 倍交易资金。这两条由程序计算数量来保证。 - 有持仓时新计划的方向必须与持仓一致；要反手先用 exit。 + +## 有持仓时 +- action=plan 只用来调整止损（只能收紧）、止盈、失效条件和 reevaluate：entries 留空，不加仓（README 第 10 节禁止计划外加仓）。止盈按当前持仓数量拆分；平仓截止时间沿用首档成交时确定的，不延长。 +- 计划仍然合适就用 keep；要提前结束用 exit（撤销未成交的入场单并市价平仓）。 +- 计划提交后由执行器在下一轮（约 2 秒内）按最新持仓核对后生效；期间持仓变了，计划作废并重新请你分析。 """ SUBMIT_SPEC = llm.ToolSpec( @@ -96,32 +99,8 @@ class Prepared: system: str user_text: str context: PlanContext - - -_exchange_info: tuple[float, dict[str, SymbolRules]] | None = None - - -async def symbol_rules(symbol: str) -> SymbolRules: - """模拟盘的价格精度、数量精度、最小下单金额，按小时缓存""" - global _exchange_info - if _exchange_info is None or time.monotonic() - _exchange_info[0] > EXCHANGE_INFO_SECONDS: - info = await public_get("/fapi/v1/exchangeInfo") - parsed = {} - for s in info["symbols"]: - filters = {f["filterType"]: f for f in s["filters"]} - parsed[s["symbol"]] = SymbolRules( - tick_size=Decimal(filters["PRICE_FILTER"]["tickSize"]).normalize(), - step_size=Decimal(filters["LOT_SIZE"]["stepSize"]).normalize(), - min_notional=Decimal(filters["MIN_NOTIONAL"]["notional"]), - ) - _exchange_info = (time.monotonic(), parsed) - if symbol not in _exchange_info[1]: - raise AnalysisError(f"模拟盘没有交易对 {symbol}") - return _exchange_info[1][symbol] - - -async def mark_price(symbol: str) -> Decimal: - return Decimal((await public_get("/fapi/v1/premiumIndex", {"symbol": symbol}))["markPrice"]) + # 当前计划的仓位计算结果，提交时重算当前止损用 + current_sizing: dict | None def plan_brief(decision: dict, sizing: dict | None) -> dict: @@ -168,9 +147,15 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared )).all() current = await session.scalar( select(TradingAnalysis) - .where(TradingAnalysis.symbol == symbol, TradingAnalysis.plan_status.in_(("pending", "open")), TradingAnalysis.dry_run.is_(False)) + .where(TradingAnalysis.symbol == symbol, TradingAnalysis.plan_status.in_(("pending", "open", "closing")), TradingAnalysis.dry_run.is_(False)) .order_by(TradingAnalysis.id.desc()).limit(1) ) + fills = [ + {"purpose": o.purpose, "leg": o.leg, "filled": str(o.filled_qty), "avg_price": str(o.avg_price), "status": o.status} + for o in (await session.scalars( + select(TradingOrder).where(TradingOrder.analysis_id == current.id, TradingOrder.filled_qty > 0).order_by(TradingOrder.id) + )).all() + ] if current else [] system = "\n\n".join([INSTRUCTIONS, "## 数据概况", await data_overview(session), await skills_section(session)]) angle, dry_run = target.angle, not account.running effort = account.reasoning_effort or provider.reasoning_effort @@ -185,9 +170,10 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared symbol_rules_ = await symbol_rules(symbol) mark = await mark_price(symbol) context = PlanContext( - angle=angle, mark_price=mark, capital=capital, other_notional=other_notional, rules=symbol_rules_, - has_plan=current is not None, position_side=("long" if position["side"] == "LONG" else "short") if position else None, + angle=angle, mark_price=mark, capital=capital, other_notional=other_notional, rules=symbol_rules_, has_plan=current is not None, + position_side=None, ) + apply_position(context, snapshot, symbol, current.sizing if current else None) orders = [o for o in snapshot["orders"] + snapshot["algo_orders"] if o["symbol"] == symbol] risk = capital * rules.RISK_PCT[angle] @@ -210,7 +196,15 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared f"- 挂单：{dump(orders) if orders else '无'}", "", "## 当前计划", - dump({"id": current.id, "plan_status": current.plan_status, **plan_brief(current.decision, current.sizing)}) if current else "无", + dump({ + "id": current.id, "plan_status": current.plan_status, + "valid_until": current.valid_until and local(current.valid_until), + "first_fill_at": current.first_fill_at and local(current.first_fill_at), + "close_deadline": current.close_deadline and local(current.close_deadline), + "current_stop": str(context.current_stop) if context.current_stop else None, + "fills": fills, + **plan_brief(current.decision, current.sizing), + }) if current else "无", "", "## 最近决策（新的在前）", *([f"- {local(a.started_at)}{'（试运行）' if a.dry_run else ''}：{dump(plan_brief(a.decision, a.sizing))}" for a in recent] or ["无"]), @@ -218,7 +212,28 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared "## 预取的行情（实盘数据，已取，不要用相同参数重复调用）", await prefetch(symbol, angle), ] - return Prepared(provider=provider, effort=effort, dry_run=dry_run, angle=angle, system=system, user_text="\n".join(lines), context=context) + return Prepared(provider=provider, effort=effort, dry_run=dry_run, angle=angle, system=system, user_text="\n".join(lines), + context=context, current_sizing=current.sizing if current else None) + + +def apply_position(context: PlanContext, snapshot: dict, symbol: str, current_sizing: dict | None) -> None: + """按快照填入持仓与当前止损。当前止损取交易所上本程序止损单里最紧的一张（可能已移到保本价），没有则取当前计划的止损""" + position = next((p for p in snapshot["positions"] if p["symbol"] == symbol), None) + if position is None: + context.position_side, context.position_qty, context.position_entry, context.current_stop = None, Decimal(0), None, None + return + side = "long" if position["side"] == "LONG" else "short" + close_side = "SELL" if side == "long" else "BUY" + stops = [ + Decimal(o["trigger_price"]) for o in snapshot["algo_orders"] + if o["symbol"] == symbol and o["side"] == close_side and o["type"].startswith("STOP") and OWN_CLIENT_ID.match(o["client_id"]) + ] + if stops: + current_stop = max(stops) if side == "long" else min(stops) + else: + current_stop = Decimal(current_sizing["stop"]) if current_sizing else None + context.position_side, context.position_qty = side, Decimal(position["amount"]) + context.position_entry, context.current_stop = Decimal(position["entry_price"]), current_stop async def complete(prepared: Prepared, messages: list[dict], tools: list[llm.ToolSpec]) -> llm.Completion: @@ -252,8 +267,11 @@ async def submit(prepared: Prepared, symbol: str, analysis_id: int, arguments: d plan = TradePlan.model_validate(arguments) except ValidationError as e: return dump({"error": "参数无效", "details": e.errors(include_url=False)}), None - # 分析可能持续几分钟，按提交时的标记价校验 + # 分析可能持续几分钟，按提交时的标记价和持仓校验 prepared.context.mark_price = await mark_price(symbol) + snapshot = await read_snapshot() + if snapshot: + apply_position(prepared.context, snapshot, symbol, prepared.current_sizing) try: sizing = size_plan(plan, prepared.context) except PlanError as e: @@ -319,7 +337,7 @@ async def run_analysis(symbol: str, reasons: list[str]) -> None: break if accepted is None: raise AnalysisError(f"模型在 {MAX_ROUNDS} 轮内没有提交通过校验的结论") - except (AnalysisError, llm.LlmError, PlanError) as e: + except (AnalysisError, ExchangeError, llm.LlmError, PlanError) as e: error = str(e) except Exception as e: logger.exception("%s 分析出错", symbol) @@ -343,6 +361,9 @@ async def run_analysis(symbol: str, reasons: list[str]) -> None: analysis.action = plan.action analysis.decision = plan.model_dump(mode="json") analysis.sizing = sizing or None + # 交易中的新计划和结束决定交给执行器按最新持仓核对后采纳 + if not dry_run and plan.action in ("plan", "exit"): + analysis.plan_status = "proposed" await session.commit() if error: await add_event(symbol, analysis_id, "analysis_failed", f"分析失败：{error}", "error") diff --git a/backend/src/alphapulse/trading/client.py b/backend/src/alphapulse/trading/client.py index 3b48589..e930582 100644 --- a/backend/src/alphapulse/trading/client.py +++ b/backend/src/alphapulse/trading/client.py @@ -7,14 +7,18 @@ import hashlib import hmac import re import time +from decimal import Decimal from typing import Any from urllib.parse import urlencode import httpx +from alphapulse.trading.plan import SymbolRules + BASE_URL = "https://demo-fapi.binance.com" RECV_WINDOW_MS = 5000 TIMEOUT_SECONDS = 10 +EXCHANGE_INFO_SECONDS = 3600 # 本程序下的单 clientOrderId / clientAlgoId 为 ap{流水 id}；其他（含用户手动下的）不动 OWN_CLIENT_ID = re.compile(r"^ap\d+$") @@ -96,3 +100,35 @@ class DemoClient: continue raise ExchangeError(resp.status_code, code, body.get("msg", resp.text[:200]) if isinstance(body, dict) else resp.text[:200]) raise AssertionError("unreachable") + + +_exchange_info: tuple[float, dict[str, SymbolRules]] | None = None + + +def step_decimal(raw: str) -> Decimal: + """去掉多余的 0（"0.10" → 0.1），但整数步长不能变成 1E+2 这类指数形式，否则按它取整的价格和数量也会是指数形式""" + value = Decimal(raw).normalize() + return value.quantize(Decimal(1)) if value.as_tuple().exponent > 0 else value + + +async def symbol_rules(symbol: str) -> SymbolRules: + """模拟盘的价格精度、数量精度、最小下单金额，按小时缓存""" + global _exchange_info + if _exchange_info is None or time.monotonic() - _exchange_info[0] > EXCHANGE_INFO_SECONDS: + info = await public_get("/fapi/v1/exchangeInfo") + parsed = {} + for s in info["symbols"]: + filters = {f["filterType"]: f for f in s["filters"]} + parsed[s["symbol"]] = SymbolRules( + tick_size=step_decimal(filters["PRICE_FILTER"]["tickSize"]), + step_size=step_decimal(filters["LOT_SIZE"]["stepSize"]), + min_notional=Decimal(filters["MIN_NOTIONAL"]["notional"]), + ) + _exchange_info = (time.monotonic(), parsed) + if symbol not in _exchange_info[1]: + raise ExchangeError(None, None, f"模拟盘没有交易对 {symbol}") + return _exchange_info[1][symbol] + + +async def mark_price(symbol: str) -> Decimal: + return Decimal((await public_get("/fapi/v1/premiumIndex", {"symbol": symbol}))["markPrice"]) diff --git a/backend/src/alphapulse/trading/engine.py b/backend/src/alphapulse/trading/engine.py index bbdf985..6e56f5b 100644 --- a/backend/src/alphapulse/trading/engine.py +++ b/backend/src/alphapulse/trading/engine.py @@ -1,4 +1,4 @@ -"""模拟盘交易进程：同步账户快照、调度分析。 +"""模拟盘交易进程：同步账户快照、执行器对账、调度分析。 用法：python -m alphapulse.trading.engine """ @@ -12,6 +12,8 @@ from sqlalchemy import select, update from alphapulse.core.db import SessionLocal from alphapulse.models import TradingAnalysis, TradingSymbol from alphapulse.trading.analyst import run_analysis +from alphapulse.trading.executor import executor_loop +from alphapulse.trading.schedule import routine_requests from alphapulse.trading.snapshot import snapshot_loop logger = logging.getLogger("alphapulse.trading.engine") @@ -57,6 +59,7 @@ async def analysis_loop() -> None: del tasks[symbol] if not task.cancelled() and task.exception(): logger.error("%s 分析任务异常退出", symbol, exc_info=task.exception()) + await routine_requests() free = MAX_CONCURRENT_ANALYSES - len(tasks) if free > 0: for symbol, reasons in await claim_requests(free, set(tasks)): @@ -69,6 +72,7 @@ async def analysis_loop() -> None: async def main() -> None: async with asyncio.TaskGroup() as tg: tg.create_task(snapshot_loop()) + tg.create_task(executor_loop()) tg.create_task(analysis_loop()) diff --git a/backend/src/alphapulse/trading/executor.py b/backend/src/alphapulse/trading/executor.py new file mode 100644 index 0000000..b018c00 --- /dev/null +++ b/backend/src/alphapulse/trading/executor.py @@ -0,0 +1,578 @@ +"""执行器：按「应有状态」对账。唯一会下单、撤单、推进计划状态的代码。 + +每轮对交易列表里的每个币： +1. 同步订单流水与交易所（成交、撤销、触发）。 +2. 采纳分析器提交的新计划（按最新持仓核对，过时则作废重算）。 +3. 推进计划状态，执行程序规则：到期、时间止损、失效条件、复核条件、无计划持仓。 +4. 算出应有订单与现有订单对比：先挂止损，再撤多余，最后补缺（撤了旧入场单的这一轮不挂新入场单）。 +""" + +import asyncio +import logging +import time +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from alphapulse.ai.tools import LOCAL_TZ +from alphapulse.core.crypto import decrypt +from alphapulse.core.db import SessionLocal +from alphapulse.market.binance import BinanceClient +from alphapulse.models import TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol +from alphapulse.trading.client import OWN_CLIENT_ID, DemoClient, ExchangeError, mark_price, symbol_rules +from alphapulse.trading.plan import MAX_HOLD_HOURS, OrderSpec, SymbolRules, condition_met, desired_orders, diff_orders, round_to_step +from alphapulse.trading.schedule import request_analysis + +logger = logging.getLogger("alphapulse.trading.executor") + +EXECUTOR_SECONDS = 2 +ACTIVE_ORDER_STATUSES = ("submitting", "new", "partially_filled") +LIVE_PLAN_STATUSES = ("pending", "open", "closing") +# 刚下的单可能短时间查不到（C1 实测约 0.25s） +SUBMIT_GRACE_SECONDS = 10 +# 止损挂单连续失败（可重试错误）达到次数后撤单平仓 +STOP_FAILURE_LIMIT = 3 +# 同一错误的时间线事件间隔，避免刷屏 +ERROR_EVENT_SECONDS = 600 +ORDER_STATUSES = {"NEW": "new", "PARTIALLY_FILLED": "partially_filled", "FILLED": "filled", "CANCELED": "canceled", + "EXPIRED": "expired", "EXPIRED_IN_MATCH": "expired", "REJECTED": "rejected"} +PURPOSE_LABELS = {"entry": "入场", "tp": "止盈", "stop": "止损", "close": "市价平仓"} +CLOSE_REASON_LABELS = {"stop": "止损", "take_profit": "止盈", "time": "到达持仓时限", "invalidation": "短线失效条件触发", + "exit": "模型决定结束", "external": "外部平仓（如在网页手动平仓）", "stop_failed": "止损挂单失败", + "close": "市价平仓"} + + +@dataclass +class Position: + side: str # long / short + qty: Decimal + entry: Decimal + break_even: Decimal + mark: Decimal + + +def fmt(value: Decimal) -> str: + return format(value, "f") + + +def utcnow() -> datetime: + return datetime.now(UTC) + + +class Executor: + def __init__(self) -> None: + self.client: DemoClient | None = None + self.key_ciphertext: bytes | None = None + self.market = BinanceClient() + # 已设置为逐仓 3 倍的币；单向持仓是账户级，只设一次 + self.configured: set[str] = set() + self.one_way_checked = False + self.stop_failures: dict[str, int] = {} + self.error_events: dict[str, float] = {} + # (symbol, 周期) -> (下一根收盘时间 ms, 最近收盘价) + self.kline_cache: dict[tuple[str, str], tuple[int, Decimal]] = {} + # 已请模型接管的无计划持仓：symbol -> 持仓均价，同一笔持仓只请求一次 + self.takeover_requested: dict[str, Decimal] = {} + + # ---------- 基础 ---------- + + async def ensure_client(self, account: TradingAccount | None) -> DemoClient | None: + ciphertext = account.api_key_encrypted if account else None + if ciphertext != self.key_ciphertext: + if self.client is not None: + await self.client.http.aclose() + self.client = None + self.configured.clear() + self.one_way_checked = False + self.key_ciphertext = ciphertext + if account is not None and ciphertext is not None: + self.client = DemoClient(decrypt(ciphertext), decrypt(account.api_secret_encrypted)) + return self.client + + async def report(self, session: AsyncSession, symbol: str | None, message: str, level: str = "error", analysis_id: int | None = None) -> None: + """同一条错误十分钟内只记一次""" + key = f"{symbol}:{message}" + if time.monotonic() - self.error_events.get(key, -ERROR_EVENT_SECONDS) < ERROR_EVENT_SECONDS: + return + self.error_events[key] = time.monotonic() + logger.warning("%s %s", symbol, message) + session.add(TradingEvent(symbol=symbol, analysis_id=analysis_id, kind="executor_error", level=level, message=message)) + + async def last_close(self, symbol: str, interval: str) -> Decimal: + """实盘该周期最近一根已收盘 K 线的收盘价；到下一根收盘前用缓存""" + now_ms = int(time.time() * 1000) + cached = self.kline_cache.get((symbol, interval)) + if cached and now_ms < cached[0]: + return cached[1] + rows = await self.market.get("um", "klines", {"symbol": symbol, "interval": interval, "limit": 2}) + closed = rows[-1] if rows[-1][6] < now_ms else rows[-2] + # 下一根收盘时间 = 当前未收盘 K 线的 close_time + next_close = rows[-1][6] + 1 if rows[-1][6] >= now_ms else now_ms + self.kline_cache[(symbol, interval)] = (next_close, Decimal(closed[4])) + return Decimal(closed[4]) + + async def condition_met(self, symbol: str, condition: dict, mark: Decimal) -> tuple[bool, Decimal]: + price = mark if condition["timeframe"] == "mark" else await self.last_close(symbol, condition["timeframe"]) + return condition_met(condition["direction"], Decimal(condition["price"]), price), price + + async def ensure_symbol_setup(self, client: DemoClient, symbol: str) -> None: + """单向持仓、逐仓、3 倍杠杆。先查再改：有挂单时即使不需要修改，写操作也可能被拒（-4067）""" + if not self.one_way_checked: + if (await client.request("GET", "/fapi/v1/positionSide/dual"))["dualSidePosition"]: + await client.request("POST", "/fapi/v1/positionSide/dual", {"dualSidePosition": False}) + self.one_way_checked = True + if symbol in self.configured: + return + config = (await client.request("GET", "/fapi/v1/symbolConfig", {"symbol": symbol}))[0] + if config["marginType"] != "ISOLATED": + await client.request("POST", "/fapi/v1/marginType", {"symbol": symbol, "marginType": "ISOLATED"}) + if config["leverage"] != 3: + await client.request("POST", "/fapi/v1/leverage", {"symbol": symbol, "leverage": 3}) + self.configured.add(symbol) + + # ---------- 订单 ---------- + + async def place(self, session: AsyncSession, client: DemoClient, symbol: str, plan_id: int | None, purpose: str, spec: OrderSpec) -> ExchangeError | None: + """先写流水并提交，再用 ap{id} 下单。结果不明（网络、5xx）时保持 submitting，下一轮按 id 查询，绝不换 id 重发""" + row = TradingOrder( + analysis_id=plan_id, symbol=symbol, purpose=purpose, leg=spec.leg, is_algo=spec.type == "STOP_MARKET", + side=spec.side, type=spec.type, price=spec.price, trigger_price=spec.trigger_price, quantity=spec.quantity, + reduce_only=purpose != "entry", status="submitting", + ) + session.add(row) + await session.commit() + client_id = f"ap{row.id}" + try: + if row.is_algo: + result = await client.request("POST", "/fapi/v1/algoOrder", { + "algoType": "CONDITIONAL", "symbol": symbol, "side": spec.side, "type": "STOP_MARKET", + "triggerPrice": fmt(spec.trigger_price), "quantity": fmt(spec.quantity), "reduceOnly": True, + "workingType": "MARK_PRICE", "clientAlgoId": client_id, + }) + row.exchange_id, row.status = str(result["algoId"]), "new" + else: + params = {"symbol": symbol, "side": spec.side, "type": spec.type, "quantity": fmt(spec.quantity), + "reduceOnly": row.reduce_only or None, "newClientOrderId": client_id} + if spec.type == "LIMIT": + params |= {"price": fmt(spec.price), "timeInForce": "GTC"} + result = await client.request("POST", "/fapi/v1/order", params) + row.exchange_id = str(result["orderId"]) + row.status = ORDER_STATUSES.get(result["status"], result["status"].lower()) + except ExchangeError as e: + row.error = str(e) + if not (e.retryable or e.status is None): + row.status = "rejected" + await session.commit() + return e + await session.commit() + return None + + async def cancel(self, client: DemoClient, row: TradingOrder) -> None: + await self.cancel_client_id(client, row.symbol, f"ap{row.id}", row.is_algo) + + async def cancel_client_id(self, client: DemoClient, symbol: str, client_id: str, is_algo: bool) -> None: + """撤单后不直接改状态：可能恰好已成交，交给下一轮同步确认""" + try: + if is_algo: + await client.request("DELETE", "/fapi/v1/algoOrder", {"clientAlgoId": client_id}) + else: + await client.request("DELETE", "/fapi/v1/order", {"symbol": symbol, "origClientOrderId": client_id}) + except ExchangeError as e: + # -2011 / -2013：单已不在，等同步 + if e.code not in (-2011, -2013): + raise + + async def sync_order(self, session: AsyncSession, client: DemoClient, row: TradingOrder, open_orders: dict, open_algos: dict) -> None: + client_id = f"ap{row.id}" + before = row.filled_qty + try: + if row.is_algo: + if client_id in open_algos: + row.exchange_id, row.status = str(open_algos[client_id]["algoId"]), "new" + return + algo = await client.request("GET", "/fapi/v1/algoOrder", {"clientAlgoId": client_id}) + if algo["algoStatus"] in ("NEW", "TRIGGERING"): + row.status = "new" + return + row.exchange_id = str(algo["algoId"]) + if algo["algoStatus"] == "FINISHED" and algo.get("actualOrderId"): + actual = await client.request("GET", "/fapi/v1/order", {"symbol": row.symbol, "orderId": algo["actualOrderId"]}) + row.filled_qty = Decimal(actual["executedQty"]) + row.avg_price = Decimal(actual["avgPrice"]) or None + row.status = "filled" if actual["status"] == "FILLED" else ORDER_STATUSES.get(actual["status"], "finished") + if row.status == "new": + # 触发后的市价单还没成交完，下一轮再查 + return + else: + row.status = algo["algoStatus"].lower() + else: + order = open_orders.get(client_id) or await client.request("GET", "/fapi/v1/order", {"symbol": row.symbol, "origClientOrderId": client_id}) + row.exchange_id = str(order["orderId"]) + row.status = ORDER_STATUSES.get(order["status"], order["status"].lower()) + row.filled_qty = Decimal(order["executedQty"]) + row.avg_price = Decimal(order["avgPrice"]) or None + except ExchangeError as e: + if e.code != -2013: + raise + if row.status == "submitting" and utcnow() - row.created_at < timedelta(seconds=SUBMIT_GRACE_SECONDS): + return + row.status, row.error = "failed", "交易所查无此单" + return + if row.filled_qty > before: + delta = row.filled_qty - before + leg = f"第 {row.leg + 1} 档" if row.purpose == "entry" else (f" {row.leg + 1} " if row.purpose == "tp" else "") + session.add(TradingEvent( + symbol=row.symbol, analysis_id=row.analysis_id, kind=f"{row.purpose}_filled", + level="warn" if row.purpose == "stop" else "info", + message=f"{PURPOSE_LABELS[row.purpose]}{leg}成交 {fmt(delta)} @ {row.avg_price}", + data={"order_id": row.id, "quantity": fmt(delta)}, + )) + if row.purpose == "tp": + await request_analysis(session, row.symbol, f"止盈 {row.leg + 1} 成交") + + # ---------- 计划 ---------- + + async def adopt(self, session: AsyncSession, client: DemoClient, symbol: TradingSymbol, current: TradingAnalysis | None, + position: Position | None) -> TradingAnalysis | None: + """采纳最新的提议；较早的提议作废。返回采纳后的当前计划""" + proposals = (await session.scalars( + select(TradingAnalysis).where(TradingAnalysis.symbol == symbol.symbol, TradingAnalysis.plan_status == "proposed") + .order_by(TradingAnalysis.id.desc()) + )).all() + if not proposals: + return current + proposal, older = proposals[0], proposals[1:] + for p in older: + p.plan_status = "stale" + now = utcnow() + if proposal.action == "exit": + proposal.plan_status = "applied" + if current is not None: + current.plan_status, current.close_reason = "closing", "exit" + session.add(TradingEvent(symbol=symbol.symbol, analysis_id=current.id, kind="plan_exit", level="warn", + message=f"按分析 #{proposal.id} 结束计划：撤销入场单，有持仓则市价平仓")) + elif position is not None: + session.add(TradingEvent(symbol=symbol.symbol, analysis_id=proposal.id, kind="plan_exit", level="warn", + message=f"按分析 #{proposal.id} 市价平掉无计划的持仓 {fmt(position.qty)}")) + close_side = "SELL" if position.side == "long" else "BUY" + await self.place(session, client, symbol.symbol, proposal.id, "close", OrderSpec("close", None, close_side, "MARKET", None, None, position.qty)) + return current + qty = position.qty if position else Decimal(0) + expected = Decimal(proposal.sizing["position_qty"]) + if qty != expected or (current is not None and current.plan_status == "closing"): + proposal.plan_status = "stale" + reason = "计划正在平仓" if current is not None and current.plan_status == "closing" else f"持仓已从 {fmt(expected)} 变为 {fmt(qty)}" + session.add(TradingEvent(symbol=symbol.symbol, analysis_id=proposal.id, kind="plan_stale", level="warn", + message=f"计划未采纳：{reason}")) + await request_analysis(session, symbol.symbol, "持仓变化，计划作废重算") + return current + if current is not None: + current.plan_status, current.close_reason, current.closed_at = "replaced", "replaced", now + if qty > 0: + proposal.plan_status = "open" + proposal.first_fill_at = current.first_fill_at if current and current.first_fill_at else now + # 持仓时限沿用原计划，不延长；接管无计划持仓时从现在起算 + proposal.close_deadline = current.close_deadline if current and current.close_deadline else now + timedelta(hours=MAX_HOLD_HOURS[symbol.angle]) + else: + proposal.plan_status = "pending" + proposal.valid_until = now + timedelta(hours=float(proposal.decision["entry_valid_hours"])) + session.add(TradingEvent(symbol=symbol.symbol, analysis_id=proposal.id, kind="plan_adopted", + message=f"采纳计划 #{proposal.id}" + (f"，替换 #{current.id}" if current else ""))) + return proposal + + async def realized(self, client: DemoClient, symbol: str, since: datetime | None) -> dict | None: + if since is None: + return None + trades = await client.request("GET", "/fapi/v1/userTrades", {"symbol": symbol, "startTime": int(since.timestamp() * 1000) - 1000, "limit": 1000}) + pnl = sum((Decimal(t["realizedPnl"]) for t in trades), Decimal(0)) + fee = sum((Decimal(t["commission"]) for t in trades if t["commissionAsset"] == "USDT"), Decimal(0)) + return {"realized_pnl": fmt(pnl), "commission": fmt(fee), "net": fmt(pnl - fee)} + + async def finish(self, session: AsyncSession, client: DemoClient, plan: TradingAnalysis, rows: list[TradingOrder]) -> None: + """持仓归零：计划结束，记录原因和盈亏，排队分析""" + if plan.close_reason is None or plan.plan_status == "open": + filled = sorted((r for r in rows if r.filled_qty > 0 and r.purpose in ("stop", "tp", "close")), key=lambda r: r.updated_at) + last = filled[-1].purpose if filled else None + plan.close_reason = {"stop": "stop", "tp": "take_profit", "close": plan.close_reason or "close"}.get(last, "external") + plan.plan_status = "closed" if plan.first_fill_at else "canceled" + plan.closed_at = utcnow() + plan.result = await self.realized(client, plan.symbol, plan.first_fill_at) + net = f"，净盈亏 {plan.result['net']} USDT（含手续费 {plan.result['commission']}）" if plan.result else "" + label = CLOSE_REASON_LABELS.get(plan.close_reason, plan.close_reason) + session.add(TradingEvent(symbol=plan.symbol, analysis_id=plan.id, kind="plan_closed", level="warn" if plan.close_reason in ("stop", "stop_failed", "external") else "info", + message=f"计划 #{plan.id} 结束：{label}{net}", data=plan.result)) + await request_analysis(session, plan.symbol, f"平仓（{label}）") + + async def start_closing(self, session: AsyncSession, plan: TradingAnalysis, reason: str, detail: str = "") -> None: + plan.plan_status, plan.close_reason = "closing", reason + session.add(TradingEvent(symbol=plan.symbol, analysis_id=plan.id, kind="plan_closing", level="warn", + message=f"{CLOSE_REASON_LABELS[reason]}{detail}：撤销入场单并市价平仓")) + + async def review_once(self, session: AsyncSession, plan_id: int, symbol: str, kind: str, index: int, message: str, reason: str) -> None: + """失效复核、reevaluate 每个条件只触发一次：以时间线事件去重""" + exists = await session.scalar(select(func.count()).where( + TradingEvent.analysis_id == plan_id, TradingEvent.kind == kind, TradingEvent.data["index"].as_integer() == index, + )) + if exists: + return + session.add(TradingEvent(symbol=symbol, analysis_id=plan_id, kind=kind, level="warn", message=message, data={"index": index})) + await request_analysis(session, symbol, reason) + + # ---------- 对账 ---------- + + async def reconcile(self, session: AsyncSession, client: DemoClient, symbol: TradingSymbol, running: bool, position: Position | None) -> None: + name = symbol.symbol + rules = await symbol_rules(name) + open_orders = {o["clientOrderId"]: o for o in await client.request("GET", "/fapi/v1/openOrders", {"symbol": name})} + open_algos = {o["clientAlgoId"]: o for o in await client.request("GET", "/fapi/v1/openAlgoOrders", {"symbol": name})} + + rows = list((await session.scalars( + select(TradingOrder).where(TradingOrder.symbol == name, TradingOrder.status.in_(ACTIVE_ORDER_STATUSES)) + )).all()) + for row in rows: + await self.sync_order(session, client, row, open_orders, open_algos) + await session.commit() + + # 交易所上本程序的单，但流水里不是活动状态：孤儿单，撤掉（用户手动下的单不动） + active_ids = {f"ap{r.id}" for r in rows if r.status in ACTIVE_ORDER_STATUSES} + for client_id, order in [*open_orders.items(), *open_algos.items()]: + if OWN_CLIENT_ID.match(client_id) and client_id not in active_ids: + await self.cancel_client_id(client, name, client_id, client_id in open_algos) + session.add(TradingEvent(symbol=name, kind="orphan_canceled", level="warn", message=f"撤销孤儿单 {client_id}（{order.get('type') or order.get('orderType')}）")) + + plan = await session.scalar( + select(TradingAnalysis).where(TradingAnalysis.symbol == name, TradingAnalysis.plan_status.in_(LIVE_PLAN_STATUSES)) + .order_by(TradingAnalysis.id.desc()).limit(1) + ) + plan = await self.adopt(session, client, symbol, plan, position) + await session.commit() + + # 不属于当前计划的活动订单（被替换、已结束的计划）：入场和止盈先撤；止损等新止损挂上后再撤，不留无保护窗口 + stale = [r for r in rows if r.status in ACTIVE_ORDER_STATUSES and r.purpose != "close" and (plan is None or r.analysis_id != plan.id)] + for row in stale: + if row.purpose != "stop": + await self.cancel(client, row) + stale_entry_canceled = any(r.purpose == "entry" for r in stale) + + if plan is None: + for row in stale: + if row.purpose == "stop": + await self.cancel(client, row) + if position is None: + self.takeover_requested.pop(name, None) + elif running: + await self.unmanaged(session, client, name, position, open_algos) + if running: + await self.check_reevaluate(session, name, position.mark if position else None) + await session.commit() + return + + direction = plan.sizing["direction"] + if position is not None and position.side != direction: + await self.report(session, name, f"持仓方向（{position.side}）与计划 #{plan.id} 方向（{direction}）不一致，暂停该币的下单", analysis_id=plan.id) + await session.commit() + return + qty = position.qty if position else Decimal(0) + now = utcnow() + plan_rows = list((await session.scalars(select(TradingOrder).where(TradingOrder.analysis_id == plan.id))).all()) + + if plan.plan_status == "pending": + if qty > 0: + plan.plan_status, plan.first_fill_at = "open", now + plan.close_deadline = now + timedelta(hours=float(plan.decision["max_hold_hours"])) + session.add(TradingEvent(symbol=name, analysis_id=plan.id, kind="plan_open", + message=f"首档成交，计划 #{plan.id} 进入持仓；最晚 {plan.close_deadline.astimezone(LOCAL_TZ).strftime('%m-%d %H:%M')} 平仓")) + await request_analysis(session, name, "首档成交") + elif now >= plan.valid_until or not running: + plan.plan_status, plan.closed_at = "canceled", now + plan.close_reason = "expired" if running else "stopped" + session.add(TradingEvent(symbol=name, analysis_id=plan.id, kind="plan_canceled", + message=f"计划 #{plan.id} 撤销：{'入场单到期未成交' if running else '已停止交易，撤销未成交的入场单'}")) + if running: + await request_analysis(session, name, "计划到期未成交") + elif plan.plan_status == "open": + if qty == 0: + await self.finish(session, client, plan, plan_rows) + elif now >= plan.close_deadline: + await self.start_closing(session, plan, "time") + else: + await self.check_invalidation(session, plan, position) + elif plan.plan_status == "closing" and qty == 0: + await self.finish(session, client, plan, plan_rows) + await session.commit() + + if plan.plan_status in LIVE_PLAN_STATUSES: + await self.sync_orders(session, client, plan, plan_rows, position, running, rules, stale_entry_canceled) + else: + # 计划已结束：只减仓止损不会被交易所自动撤销，必须主动撤，以免作用到之后的新仓位 + for row in plan_rows: + if row.status in ACTIVE_ORDER_STATUSES and row.purpose != "close": + await self.cancel(client, row) + for row in stale: + if row.purpose == "stop": + await self.cancel(client, row) + if running and plan.plan_status != "closing": + await self.check_reevaluate(session, name, position.mark if position else None) + await session.commit() + + async def check_invalidation(self, session: AsyncSession, plan: TradingAnalysis, position: Position) -> None: + condition = plan.decision["invalidation"] + met, price = await self.condition_met(plan.symbol, condition, position.mark) + if not met: + return + detail = f"（{condition['timeframe']} 收盘 {fmt(price)} {'高于' if condition['direction'] == 'above' else '低于'} {condition['price']}）" + if plan.angle == "short": + await self.start_closing(session, plan, "invalidation", detail) + else: + await self.review_once(session, plan.id, plan.symbol, "invalidation_review", 0, + f"长线失效条件触发{detail}，请模型复核", f"失效条件触发{detail}") + + async def check_reevaluate(self, session: AsyncSession, symbol: str, mark: Decimal | None) -> None: + """最近一次交易中分析给出的 reevaluate 条件：满足任一个就排队分析""" + latest = await session.scalar( + select(TradingAnalysis).where(TradingAnalysis.symbol == symbol, TradingAnalysis.status == "done", TradingAnalysis.dry_run.is_(False)) + .order_by(TradingAnalysis.id.desc()).limit(1) + ) + if latest is None or not latest.decision["reevaluate"]: + return + if mark is None and any(c["timeframe"] == "mark" for c in latest.decision["reevaluate"]): + mark = await mark_price(symbol) + for i, condition in enumerate(latest.decision["reevaluate"]): + met, price = await self.condition_met(symbol, condition, mark) + if met: + text = f"{condition['timeframe']} {'高于' if condition['direction'] == 'above' else '低于'} {condition['price']}（现 {fmt(price)}）" + await self.review_once(session, latest.id, symbol, "reevaluate_triggered", i, f"复核条件满足：{text}", f"复核条件满足：{text}") + + async def unmanaged(self, session: AsyncSession, client: DemoClient, symbol: str, position: Position, open_algos: dict) -> None: + """有持仓但没有计划：有止损就请模型接管，没有止损就市价平仓""" + close_side = "SELL" if position.side == "long" else "BUY" + has_stop = any(o["side"] == close_side and o["orderType"].startswith("STOP") for o in open_algos.values()) + if has_stop: + if self.takeover_requested.get(symbol) != position.entry: + self.takeover_requested[symbol] = position.entry + session.add(TradingEvent(symbol=symbol, kind="unmanaged_position", level="warn", + message=f"{symbol} 有持仓 {fmt(position.qty)} 和止损但没有计划，请模型接管")) + await request_analysis(session, symbol, "有持仓但没有计划，请接管") + return + closing = await session.scalar(select(func.count()).where( + TradingOrder.symbol == symbol, TradingOrder.purpose == "close", TradingOrder.status.in_(ACTIVE_ORDER_STATUSES), + )) + if closing: + return + session.add(TradingEvent(symbol=symbol, kind="unmanaged_closed", level="error", + message=f"{symbol} 有持仓 {fmt(position.qty)} 但没有计划也没有止损，市价平仓")) + await self.place(session, client, symbol, None, "close", OrderSpec("close", None, close_side, "MARKET", None, None, position.qty)) + + async def sync_orders(self, session: AsyncSession, client: DemoClient, plan: TradingAnalysis, plan_rows: list[TradingOrder], + position: Position | None, running: bool, rules: SymbolRules, block_entries: bool) -> None: + sizing = plan.sizing + qty = position.qty if position else Decimal(0) + n_entries, n_tps = len(sizing["entries"]), len(sizing["take_profits"]) + entry_filled = [sum((r.filled_qty for r in plan_rows if r.purpose == "entry" and r.leg == i), Decimal(0)) for i in range(n_entries)] + tp_filled = [sum((r.filled_qty for r in plan_rows if r.purpose == "tp" and r.leg == j), Decimal(0)) for j in range(n_tps)] + tp1_hit = tp_filled[0] > 0 + stop = Decimal(sizing["stop"]) + if tp1_hit and position is not None: + # 止盈 1 后止损移到保本价：做多向上取整、做空向下取整，保证不亏；只收紧不放宽 + long = sizing["direction"] == "long" + break_even = round_to_step(position.break_even, rules.tick_size, ROUND_CEILING if long else ROUND_FLOOR) + stop = max(stop, break_even) if long else min(stop, break_even) + + active = [r for r in plan_rows if r.status in ACTIVE_ORDER_STATUSES and r.purpose in ("entry", "stop", "tp")] + long = sizing["direction"] == "long" + if plan.plan_status == "closing": + # 平仓中：撤入场和止盈，保留已有止损直到市价平仓成交（止损可能正因越过现价挂不上，不再补挂） + for row in active: + if row.purpose != "stop": + await self.cancel(client, row) + if qty > 0 and not any(r.purpose == "close" and r.status in ACTIVE_ORDER_STATUSES for r in plan_rows): + close_side = "SELL" if long else "BUY" + await self.place(session, client, plan.symbol, plan.id, "close", OrderSpec("close", None, close_side, "MARKET", None, None, qty)) + return + + entries_open = running and utcnow() < (plan.valid_until or utcnow()) and not tp1_hit + desired = desired_orders(sizing, entries_open=entries_open, entry_filled=entry_filled, tp_filled=tp_filled, + position_qty=qty, stop_price=stop, step_size=rules.step_size) + actual = [(r, OrderSpec(r.purpose, r.leg, r.side, r.type, r.price, r.trigger_price, r.quantity - r.filled_qty)) for r in active] + to_cancel, to_place = diff_orders(desired, actual) + + entries = [d for d in to_place if d.purpose == "entry"] + if entries: + # 保证金模式和杠杆在有挂单时不能改，必须在挂任何单之前设好 + await self.ensure_symbol_setup(client, plan.symbol) + # 1. 止损先挂新的再撤旧的（只减仓止损可以并存），任何时候都不留无保护窗口 + for spec in [d for d in to_place if d.purpose == "stop"]: + error = await self.place(session, client, plan.symbol, plan.id, "stop", spec) + if error is not None: + await self.stop_failed(session, plan, error) + return + self.stop_failures.pop(plan.symbol, None) + # 2. 撤多余的单 + canceled_entry = False + for row in to_cancel: + await self.cancel(client, row) + canceled_entry = canceled_entry or row.purpose == "entry" + # 3. 补止盈；入场单只在本轮没有撤旧入场单、且止损在场时挂 + has_stop = any(r.purpose == "stop" for r in active if r not in to_cancel) or any(d.purpose == "stop" for d in to_place) + for spec in [d for d in to_place if d.purpose == "tp"]: + await self.place(session, client, plan.symbol, plan.id, "tp", spec) + if entries and has_stop and not canceled_entry and not block_entries: + for spec in entries: + await self.place(session, client, plan.symbol, plan.id, "entry", spec) + + async def stop_failed(self, session: AsyncSession, plan: TradingAnalysis, error: ExchangeError) -> None: + """会立即触发（-2021）或不可重试的错误：撤单平仓；可重试的错误连续失败多次后同样处理""" + if error.retryable or error.status is None: + count = self.stop_failures[plan.symbol] = self.stop_failures.get(plan.symbol, 0) + 1 + if count < STOP_FAILURE_LIMIT: + await self.report(session, plan.symbol, f"挂止损失败（第 {count} 次），下一轮重试：{error}", "warn", plan.id) + return + self.stop_failures.pop(plan.symbol, None) + reason = "止损价已越过现价" if error.code == -2021 else f"挂止损失败：{error}" + session.add(TradingEvent(symbol=plan.symbol, analysis_id=plan.id, kind="stop_failed", level="error", message=reason)) + await self.start_closing(session, plan, "stop_failed") + + # ---------- 主循环 ---------- + + async def tick(self) -> None: + async with SessionLocal() as session: + account = await session.get(TradingAccount, 1) + symbols = list((await session.scalars(select(TradingSymbol))).all()) + client = await self.ensure_client(account) + if client is None or not symbols: + return + positions = {} + for p in await client.request("GET", "/fapi/v3/positionRisk"): + amount = Decimal(p["positionAmt"]) + if amount != 0: + positions[p["symbol"]] = Position( + side="long" if amount > 0 else "short", qty=abs(amount), entry=Decimal(p["entryPrice"]), + break_even=Decimal(p["breakEvenPrice"]), mark=Decimal(p["markPrice"]), + ) + for symbol in symbols: + try: + await self.reconcile(session, client, symbol, account.running, positions.get(symbol.symbol)) + except ExchangeError as e: + await session.rollback() + await self.report(session, symbol.symbol, f"对账失败：{e}") + await session.commit() + except Exception as e: + logger.exception("%s 对账出错", symbol.symbol) + await session.rollback() + await self.report(session, symbol.symbol, f"对账出错：{e!r}") + await session.commit() + + +async def executor_loop() -> None: + executor = Executor() + while True: + try: + await executor.tick() + except ExchangeError as e: + logger.warning("执行器同步失败：%s", e) + except Exception: + logger.exception("执行器出错，稍后重试") + await asyncio.sleep(EXECUTOR_SECONDS) diff --git a/backend/src/alphapulse/trading/plan.py b/backend/src/alphapulse/trading/plan.py index feed357..fdad159 100644 --- a/backend/src/alphapulse/trading/plan.py +++ b/backend/src/alphapulse/trading/plan.py @@ -4,6 +4,7 @@ """ from dataclasses import dataclass +from datetime import UTC, datetime, timedelta from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal from typing import Annotated, Literal @@ -32,6 +33,10 @@ TAKER_FEE_RATE = Decimal("0.0004") # 模拟盘 PERCENT_PRICE：买单价不高于标记价 × 1.05，卖单价不低于标记价 × 0.95 PERCENT_PRICE_LIMIT = Decimal("0.05") RATIO_TOLERANCE = Decimal("0.001") +# 例行分析周期（小时）：短线跟 1h、长线跟 4h 收盘 +ROUTINE_HOURS = {"short": 1, "long": 4} +# 同一个币两次自动分析的最短间隔 +MIN_ANALYSIS_GAP = timedelta(minutes=10) Price = Annotated[Decimal, Field(gt=0)] @@ -99,6 +104,11 @@ class PlanContext: rules: SymbolRules has_plan: bool position_side: Direction | None + position_qty: Decimal = Decimal(0) + # 持仓均价 + position_entry: Decimal | None = None + # 当前生效的止损价（可能已移到保本价）；新计划只能收紧 + current_stop: Decimal | None = None class PlanError(Exception): @@ -142,14 +152,27 @@ def check_action(plan: TradePlan, ctx: PlanContext) -> list[str]: def check_plan_fields(plan: TradePlan, ctx: PlanContext) -> list[str]: - """action=plan 时的字段完整性、档数、比例、有效期""" - required = ("direction", "confidence", "entries", "stop", "take_profits", "entry_valid_hours", "max_hold_hours", "invalidation") + """action=plan 时的字段完整性、档数、比例、有效期。有持仓时只调整止损止盈等，不加仓""" + holding = ctx.position_side is not None + required = ("direction", "confidence", "stop", "take_profits", "invalidation") + if not holding: + required += ("entries", "entry_valid_hours", "max_hold_hours") missing = [name for name in required if getattr(plan, name) is None] if missing: return [f"action=plan 时缺少：{', '.join(missing)}"] errors = [] angle = ctx.angle label = "短线" if angle == "short" else "长线" + if holding: + if plan.entries: + errors.append("已有持仓，新计划不加仓：entries 须为空，只调整止损、止盈、失效条件和 reevaluate") + if plan.direction != ctx.position_side: + errors.append(f"已有{'多' if ctx.position_side == 'long' else '空'}头持仓，新计划方向必须一致；要反手请先用 exit") + if len(plan.take_profits) != len(TAKE_PROFIT_RATIOS[angle]): + errors.append(f"{label}须给 {len(TAKE_PROFIT_RATIOS[angle])} 个止盈，现为 {len(plan.take_profits)}") + if plan.invalidation.timeframe not in INVALIDATION_TIMEFRAMES[angle]: + errors.append(f"{label}失效条件的周期只能是 {'/'.join(INVALIDATION_TIMEFRAMES[angle])}") + return errors low, high = ENTRY_COUNT[angle] if not low <= len(plan.entries) <= high: errors.append(f"{label}入场档数须为 {low}–{high}，现为 {len(plan.entries)}" if low != high else f"{label}只能 {low} 档入场，现为 {len(plan.entries)}") @@ -163,8 +186,6 @@ def check_plan_fields(plan: TradePlan, ctx: PlanContext) -> list[str]: errors.append(f"{label} max_hold_hours 不能超过 {MAX_HOLD_HOURS[angle]}") if plan.invalidation.timeframe not in INVALIDATION_TIMEFRAMES[angle]: errors.append(f"{label}失效条件的周期只能是 {'/'.join(INVALIDATION_TIMEFRAMES[angle])}") - if ctx.position_side is not None and plan.direction != ctx.position_side: - errors.append(f"已有{'多' if ctx.position_side == 'long' else '空'}头持仓，新计划方向必须一致；要反手请先用 exit") return errors @@ -214,6 +235,8 @@ def size_plan(plan: TradePlan, ctx: PlanContext) -> dict: errors += check_plan_fields(plan, ctx) if errors: raise PlanError(errors) + if ctx.position_side is not None: + return size_managed(plan, ctx) angle, direction, rules = ctx.angle, plan.direction, ctx.rules # 向保守方向取整：做多的入场、止损、止盈都向下（入场更低、止损更远、止盈更近），做空都向上 @@ -282,4 +305,138 @@ def size_plan(plan: TradePlan, ctx: PlanContext) -> dict: "capped_by_leverage": by_leverage < by_risk, "reward_risk": str(rr.quantize(Decimal("0.01"))), "reward_risk_first_only": str(rr_first.quantize(Decimal("0.01"))), + # 执行器采纳前核对：持仓变了说明计划基于过时状态 + "position_qty": "0", + "managed": False, } + + +def size_managed(plan: TradePlan, ctx: PlanContext) -> dict: + """有持仓时的新计划：不加仓，止损只能收紧，止盈按当前持仓数量拆分""" + direction, rules = plan.direction, ctx.rules + rounding = ROUND_FLOOR if direction == "long" else ROUND_CEILING + stop = round_to_step(plan.stop, rules.tick_size, rounding) + targets = [round_to_step(t.price, rules.tick_size, rounding) for t in plan.take_profits] + sign = 1 if direction == "long" else -1 + mark, invalidation = ctx.mark_price, plan.invalidation.price + + def below(a: Decimal, b: Decimal) -> bool: + return sign * (b - a) > 0 + + side = "做多" if direction == "long" else "做空" + errors = [] + if plan.invalidation.direction != ("below" if direction == "long" else "above"): + errors.append(f"{side}的失效条件方向应为 {'below' if direction == 'long' else 'above'}") + if not (below(stop, invalidation) and below(invalidation, mark)): + errors.append(f"{side}时须满足 止损 {stop} → 失效价 {invalidation} → 现价 {mark} 的顺序") + if ctx.current_stop is not None and below(stop, ctx.current_stop): + errors.append(f"止损只能收紧：当前止损 {ctx.current_stop}，新止损 {stop} 更宽") + if targets and not below(mark, targets[0]): + errors.append(f"止盈 1 {targets[0]} 已越过现价 {mark}，挂单会立即成交") + if any(not below(a, b) for a, b in zip(targets, targets[1:], strict=False)): + errors.append("止盈价须按由近到远排列且不重复") + if errors: + raise PlanError(errors) + + tp_ratios = TAKE_PROFIT_RATIOS[ctx.angle] + entry = ctx.position_entry + quantity = ctx.position_qty + rr = reward_risk(mark, stop, targets, tp_ratios) if ctx.angle == "long" else reward_risk(mark, stop, targets[:1], (Decimal(1),)) + loss = quantity * (sign * (entry - stop) + (entry + stop) * TAKER_FEE_RATE) + return { + "direction": direction, + "entries": [], + "stop": str(stop), + "take_profits": [ + {"price": str(t), "ratio": str(r), "quantity": str(q)} + for t, r, q in zip(targets, tp_ratios, split_take_profits(quantity, tp_ratios, rules.step_size), strict=True) + ], + "average_price": str(entry), + "quantity": str(quantity), + "notional": str((entry * quantity).quantize(Decimal("0.01"))), + "capital": str(ctx.capital), + "leverage": str((entry * quantity / ctx.capital).quantize(Decimal("0.01"))), + "risk_amount": "0", + # 止损已越过成本价时为负，表示锁定盈利 + "risk_at_stop": str(loss.quantize(Decimal("0.01"))), + "risk_pct": str((loss / ctx.capital * 100).quantize(Decimal("0.01"))), + "capped_by_leverage": False, + # 按现价计算 + "reward_risk": str(rr.quantize(Decimal("0.01"))), + "reward_risk_first_only": str(rr.quantize(Decimal("0.01"))), + "position_qty": str(quantity), + "managed": True, + } + + +# ---------- 执行：应有订单 ---------- + +@dataclass(frozen=True) +class OrderSpec: + """一张应有（或现有）的订单；quantity 为剩余未成交数量""" + + purpose: Literal["entry", "stop", "tp", "close"] + leg: int | None + side: Literal["BUY", "SELL"] + type: Literal["LIMIT", "STOP_MARKET", "MARKET"] + price: Decimal | None + trigger_price: Decimal | None + quantity: Decimal + + +def desired_orders( + sizing: dict, *, entries_open: bool, entry_filled: list[Decimal], tp_filled: list[Decimal], + position_qty: Decimal, stop_price: Decimal, step_size: Decimal, +) -> list[OrderSpec]: + """计划在当前成交状态下应挂的单。 + + - 入场：entries_open 时各档剩余数量。 + - 止损：有持仓或有入场单时一张只减仓止损，数量取计划总量与持仓的较大者（数量大于持仓时交易所只平实际持仓）。 + - 止盈：有持仓时按累计成交数量拆分，减去各档已成交。 + """ + long = sizing["direction"] == "long" + open_side, close_side = ("BUY", "SELL") if long else ("SELL", "BUY") + orders = [] + if entries_open: + for i, e in enumerate(sizing["entries"]): + remaining = Decimal(e["quantity"]) - entry_filled[i] + if remaining > 0: + orders.append(OrderSpec("entry", i, open_side, "LIMIT", Decimal(e["price"]), None, remaining)) + if position_qty > 0 or orders: + quantity = max(Decimal(sizing["quantity"]), position_qty) + orders.append(OrderSpec("stop", None, close_side, "STOP_MARKET", None, stop_price, quantity)) + if position_qty > 0: + filled_total = Decimal(sizing["quantity"]) if sizing["managed"] else sum(entry_filled, Decimal(0)) + ratios = tuple(Decimal(t["ratio"]) for t in sizing["take_profits"]) + for j, (t, target) in enumerate(zip(sizing["take_profits"], split_take_profits(filled_total, ratios, step_size), strict=True)): + remaining = target - tp_filled[j] + if remaining > 0: + orders.append(OrderSpec("tp", j, close_side, "LIMIT", Decimal(t["price"]), None, remaining)) + return orders + + +def diff_orders[K](desired: list[OrderSpec], actual: list[tuple[K, OrderSpec]]) -> tuple[list[K], list[OrderSpec]]: + """对比应有与现有订单：返回（要撤销的现有订单，要新挂的订单）。价格、数量完全一致才算同一张""" + missing = list(desired) + cancel = [] + for key, spec in actual: + if spec in missing: + missing.remove(spec) + else: + cancel.append(key) + return cancel, missing + + +def condition_met(direction: Literal["above", "below"], threshold: Decimal, price: Decimal) -> bool: + return price > threshold if direction == "above" else price < threshold + + +def next_routine_at(last_started: datetime, angle: Angle) -> datetime: + """例行分析时间：上次分析后约 3/4 个周期之后的第一个 K 线收盘，再加 1 分钟等收盘数据就绪。 + 任何一次分析（包括触发式）都会让例行分析重新计时""" + period = timedelta(hours=ROUTINE_HOURS[angle]) + # 币安 K 线按 UTC 对齐 + earliest = (last_started + period * 3 / 4).astimezone(UTC) + day = earliest.replace(hour=0, minute=0, second=0, microsecond=0) + boundary = day + period * -(-(earliest - day) // period) + return boundary + timedelta(minutes=1) diff --git a/backend/src/alphapulse/trading/schedule.py b/backend/src/alphapulse/trading/schedule.py new file mode 100644 index 0000000..87c35b8 --- /dev/null +++ b/backend/src/alphapulse/trading/schedule.py @@ -0,0 +1,68 @@ +"""自动分析的排队：触发式（成交、止盈、平仓、到期、条件满足）和例行。只在交易中排队，手动分析走 API""" + +from datetime import UTC, datetime, timedelta + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from alphapulse.ai.tools import LOCAL_TZ +from alphapulse.core.db import SessionLocal +from alphapulse.models import TradingAccount, TradingAnalysis, TradingEvent, TradingSymbol +from alphapulse.trading.plan import DAILY_ANALYSIS_LIMIT, MIN_ANALYSIS_GAP, next_routine_at + +# 自动分析失败后的重试：间隔、连续失败上限 +RETRY_AFTER = timedelta(minutes=5) +MAX_CONSECUTIVE_FAILURES = 3 + + +async def daily_limit_reached(session: AsyncSession) -> bool: + today = datetime.now(LOCAL_TZ).replace(hour=0, minute=0, second=0, microsecond=0) + count = await session.scalar(select(func.count()).where(TradingAnalysis.started_at >= today)) + if count < DAILY_ANALYSIS_LIMIT: + return False + noticed = await session.scalar(select(func.count()).where(TradingEvent.kind == "daily_limit", TradingEvent.created_at >= today)) + if not noticed: + session.add(TradingEvent(kind="daily_limit", level="warn", + message=f"今日自动分析已达 {DAILY_ANALYSIS_LIMIT} 次上限，之后只执行程序规则（止损、止盈、到期平仓等）")) + return True + + +async def request_analysis(session: AsyncSession, symbol: str, reason: str) -> None: + """排队一次自动分析：多个原因合并；距上次分析不足最短间隔时顺延。调用方负责 commit""" + account = await session.get(TradingAccount, 1) + target = await session.get(TradingSymbol, symbol) + if account is None or not account.running or target is None or await daily_limit_reached(session): + return + now = datetime.now(UTC) + last_started = await session.scalar(select(func.max(TradingAnalysis.started_at)).where(TradingAnalysis.symbol == symbol)) + at = max(now, last_started + MIN_ANALYSIS_GAP) if last_started else now + target.analysis_requested_at = min(target.analysis_requested_at, at) if target.analysis_requested_at else at + if reason not in target.analysis_reasons: + # JSONB 列要整体赋值才会被识别为修改 + target.analysis_reasons = [*target.analysis_reasons, reason] + + +async def routine_requests() -> None: + """交易中：到例行时间的币排队例行分析；自动分析失败的隔一段时间重试""" + async with SessionLocal() as session: + account = await session.get(TradingAccount, 1) + if account is None or not account.running: + return + now = datetime.now(UTC) + for target in (await session.scalars(select(TradingSymbol))).all(): + if target.analysis_requested_at is not None: + continue + recent = (await session.scalars( + select(TradingAnalysis).where(TradingAnalysis.symbol == target.symbol) + .order_by(TradingAnalysis.id.desc()).limit(MAX_CONSECUTIVE_FAILURES) + )).all() + if recent and recent[0].status == "running": + continue + if not recent or now >= next_routine_at(recent[0].started_at, target.angle): + await request_analysis(session, target.symbol, "例行分析") + continue + last = recent[0] + all_failed = len(recent) == MAX_CONSECUTIVE_FAILURES and all(a.status == "failed" for a in recent) + if last.status == "failed" and last.reasons != ["手动分析"] and not all_failed and now >= last.finished_at + RETRY_AFTER: + await request_analysis(session, target.symbol, f"重试：{'；'.join(last.reasons)}") + await session.commit() diff --git a/backend/tests/test_plan.py b/backend/tests/test_plan.py index 02c8fca..7baee34 100644 --- a/backend/tests/test_plan.py +++ b/backend/tests/test_plan.py @@ -1,8 +1,12 @@ +from datetime import UTC, datetime, timedelta, timezone from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal import pytest -from alphapulse.trading.plan import PlanContext, PlanError, SymbolRules, TradePlan, round_to_step, size_plan, split_take_profits +from alphapulse.trading.plan import ( + OrderSpec, PlanContext, PlanError, SymbolRules, TradePlan, desired_orders, diff_orders, next_routine_at, round_to_step, + size_plan, split_take_profits, +) BTC = SymbolRules(tick_size=Decimal("0.1"), step_size=Decimal("0.0001"), min_notional=Decimal("50")) @@ -137,3 +141,99 @@ def test_action_rules(): reevaluate=[{"direction": "above", "price": "81000", "timeframe": "1h", "basis": "x"}]) assert "已经满足" in errors_of(satisfied, context()) assert "方向必须一致" in errors_of(long_plan(), context(position_side="short")) + + +# ---------- 有持仓时的新计划 ---------- + +def managed_plan(**overrides) -> TradePlan: + values = { + "action": "plan", "summary": "调整", "reasons": "r", "risks": "x", "direction": "long", "confidence": "medium", + "stop": "80000", "take_profits": [{"price": "83000", "basis": "a"}, {"price": "84000", "basis": "b"}, {"price": "85000", "basis": "c"}], + "invalidation": {"direction": "below", "price": "80500", "timeframe": "4h", "basis": "x"}, + } + return TradePlan.model_validate({**values, **overrides}) + + +def holding(**overrides) -> PlanContext: + return context(position_side="long", position_qty=Decimal("0.0100"), position_entry=Decimal("81000"), current_stop=Decimal("79500"), has_plan=True, **overrides) + + +def test_managed_plan_does_not_add_and_splits_position(): + sizing = size_plan(managed_plan(), holding()) + assert sizing["managed"] is True + assert sizing["entries"] == [] + assert [t["quantity"] for t in sizing["take_profits"]] == ["0.0050", "0.0030", "0.0020"] + assert sizing["position_qty"] == "0.0100" + assert "entries 须为空" in errors_of(managed_plan(entries=[{"price": "80000", "ratio": "1", "basis": "a"}]), holding()) + + +def test_managed_plan_only_tightens_stop(): + assert "止损只能收紧" in errors_of(managed_plan(stop="79000", invalidation={"direction": "below", "price": "79400", "timeframe": "4h", "basis": "x"}), holding()) + + +# ---------- 应有订单 ---------- + +LONG_SIZING = { + "direction": "long", "managed": False, "quantity": "0.0030", + "entries": [{"price": "81000.0", "quantity": "0.0012"}, {"price": "80000.0", "quantity": "0.0018"}], + "take_profits": [{"price": "83000.0", "ratio": "0.5"}, {"price": "84000.0", "ratio": "0.3"}, {"price": "85000.0", "ratio": "0.2"}], +} +ZERO3 = [Decimal(0)] * 3 + + +def test_desired_orders_before_fill_places_entries_and_stop_together(): + orders = desired_orders(LONG_SIZING, entries_open=True, entry_filled=[Decimal(0)] * 2, tp_filled=ZERO3, + position_qty=Decimal(0), stop_price=Decimal("79500"), step_size=Decimal("0.0001")) + assert [(o.purpose, o.leg, o.side, o.quantity) for o in orders] == [ + ("entry", 0, "BUY", Decimal("0.0012")), ("entry", 1, "BUY", Decimal("0.0018")), ("stop", None, "SELL", Decimal("0.0030")), + ] + + +def test_desired_orders_partial_fill_splits_take_profits_by_filled(): + orders = desired_orders(LONG_SIZING, entries_open=True, entry_filled=[Decimal("0.0012"), Decimal("0.0004")], tp_filled=ZERO3, + position_qty=Decimal("0.0016"), stop_price=Decimal("79500"), step_size=Decimal("0.0001")) + assert [(o.purpose, o.leg, o.quantity) for o in orders] == [ + ("entry", 1, Decimal("0.0014")), ("stop", None, Decimal("0.0030")), + ("tp", 0, Decimal("0.0008")), ("tp", 1, Decimal("0.0004")), ("tp", 2, Decimal("0.0004")), + ] + + +def test_desired_orders_after_tp1_keeps_remaining_targets_only(): + orders = desired_orders(LONG_SIZING, entries_open=False, entry_filled=[Decimal("0.0012"), Decimal("0.0018")], + tp_filled=[Decimal("0.0015"), Decimal(0), Decimal(0)], position_qty=Decimal("0.0015"), + stop_price=Decimal("81050"), step_size=Decimal("0.0001")) + assert [(o.purpose, o.leg, o.quantity, o.trigger_price) for o in orders] == [ + ("stop", None, Decimal("0.0030"), Decimal("81050")), ("tp", 1, Decimal("0.0009"), None), ("tp", 2, Decimal("0.0006"), None), + ] + + +def test_desired_orders_nothing_when_flat_and_entries_closed(): + assert desired_orders(LONG_SIZING, entries_open=False, entry_filled=[Decimal(0)] * 2, tp_filled=ZERO3, + position_qty=Decimal(0), stop_price=Decimal("79500"), step_size=Decimal("0.0001")) == [] + + +def test_diff_orders_matches_by_value(): + keep = OrderSpec("stop", None, "SELL", "STOP_MARKET", None, Decimal("79500.0"), Decimal("0.0030")) + stale = OrderSpec("entry", 0, "BUY", "LIMIT", Decimal("81000"), None, Decimal("0.0012")) + wanted = OrderSpec("entry", 0, "BUY", "LIMIT", Decimal("80900"), None, Decimal("0.0012")) + cancel, place = diff_orders([OrderSpec("stop", None, "SELL", "STOP_MARKET", None, Decimal("79500"), Decimal("0.003")), wanted], + [(1, keep), (2, stale)]) + assert cancel == [2] + assert place == [wanted] + + +def test_next_routine_at_aligns_to_candle_close(): + tz8 = timezone(timedelta(hours=8)) + # 北京 10:20 = UTC 02:20；短线 +45 分钟后的第一个整点 UTC 04:00，再加 1 分钟 + assert next_routine_at(datetime(2026, 10, 15, 10, 20, tzinfo=tz8), "short") == datetime(2026, 10, 15, 4, 1, tzinfo=UTC) + assert next_routine_at(datetime(2026, 10, 15, 3, 1, tzinfo=UTC), "short") == datetime(2026, 10, 15, 4, 1, tzinfo=UTC) + # 长线：UTC 04:01 + 3h = 07:01，下一个 4h 收盘 08:00 + assert next_routine_at(datetime(2026, 10, 15, 4, 1, tzinfo=UTC), "long") == datetime(2026, 10, 15, 8, 1, tzinfo=UTC) + + +def test_step_decimal_never_uses_exponent(): + from alphapulse.trading.client import step_decimal + + assert str(step_decimal("0.10")) == "0.1" + assert str(step_decimal("100")) == "100" + assert str(round_to_step(Decimal("12345"), step_decimal("100"), ROUND_FLOOR)) == "12300" diff --git a/web/src/apis/trading.ts b/web/src/apis/trading.ts index d48b08c..4a64152 100644 --- a/web/src/apis/trading.ts +++ b/web/src/apis/trading.ts @@ -77,7 +77,10 @@ export interface TradingSymbol { added_at: string analysis_requested_at: string | null analyzing: boolean - plan_status: "pending" | "open" | null + /** 执行器正在管理的计划 */ + plan_id: number | null + plan_status: "pending" | "open" | "closing" | null + close_deadline: string | null } export interface SymbolOption { @@ -183,6 +186,10 @@ export interface Sizing { capped_by_leverage: boolean reward_risk: string reward_risk_first_only: string + /** 采纳前核对用：提交时的持仓数量 */ + position_qty: string + /** 持仓中调整的计划：不加仓 */ + managed: boolean } export interface Analysis { @@ -201,13 +208,13 @@ export interface Analysis { usage: { input_tokens: number, output_tokens: number } | null started_at: string finished_at: string | null - plan_status: "pending" | "open" | "closed" | "canceled" | "replaced" | null + plan_status: PlanStatus | null valid_until: string | null first_fill_at: string | null close_deadline: string | null closed_at: string | null close_reason: string | null - result: Record | null + result: { realized_pnl: string, commission: string, net: string } | null } /** 与协议无关的消息格式，同 Chat */ @@ -221,6 +228,41 @@ export interface AnalysisDetail extends Analysis { transcript: TranscriptMessage[] | null } +export type PlanStatus = "proposed" | "pending" | "open" | "closing" | "closed" | "canceled" | "replaced" | "stale" | "applied" + +export const PLAN_STATUS_LABELS: Record = { + proposed: "待采纳", + pending: "挂单中", + open: "持仓中", + closing: "平仓中", + closed: "已结束", + canceled: "已撤销", + replaced: "已被替换", + stale: "已作废", + applied: "已执行" +} + +/** 执行器的订单流水 */ +export interface OrderRow { + id: number + purpose: "entry" | "stop" | "tp" | "close" + leg: number | null + side: "BUY" | "SELL" + type: string + price: string | null + trigger_price: string | null + quantity: string + status: string + filled_qty: string + avg_price: string | null + error: string | null + created_at: string +} + +export function getSymbolPlan(symbol: string): Promise<{ plan: Analysis | null, orders: OrderRow[] }> { + return http.get(`/trading/symbols/${symbol}/plan`) +} + export function requestAnalysis(symbol: string): Promise { return http.post(`/trading/symbols/${symbol}/analyze`) } diff --git a/web/src/pages/trading/components/PlanCard.vue b/web/src/pages/trading/components/PlanCard.vue index d7049ef..35ee89d 100644 --- a/web/src/pages/trading/components/PlanCard.vue +++ b/web/src/pages/trading/components/PlanCard.vue @@ -1,9 +1,10 @@