From 005943a54c0c41d9aae6ba53ec0d387c8189a6fe Mon Sep 17 00:00:00 2001 From: eezd Date: Thu, 15 Oct 2026 09:54:47 +0800 Subject: [PATCH] Add demo trading page, trader snapshot service and trading tables Add trading_symbols, trading_analyses, trading_orders and trading_events. The new trader service publishes account, position and open-order snapshots of the Binance demo account to Redis every 3s. The /trading page shows engine status, start/stop, balances, positions, the symbol list with angle selection, open orders and the event timeline. --- .../20261015_a54081406f23_trading_tables.py | 114 +++++ backend/src/alphapulse/api/main.py | 3 +- backend/src/alphapulse/api/trading.py | 271 ++++++++++++ backend/src/alphapulse/models/__init__.py | 6 +- backend/src/alphapulse/models/trading.py | 105 ++++- backend/src/alphapulse/trading/client.py | 12 + backend/src/alphapulse/trading/engine.py | 159 +++++++ deploy/docker-compose.yml | 4 + web/src/apis/trading.ts | 138 ++++++ .../trading/components/AddSymbolDialog.vue | 70 +++ web/src/pages/trading/index.vue | 409 ++++++++++++++++++ web/src/router/index.ts | 1 + 12 files changed, 1289 insertions(+), 3 deletions(-) create mode 100644 backend/alembic/versions/20261015_a54081406f23_trading_tables.py create mode 100644 backend/src/alphapulse/api/trading.py create mode 100644 backend/src/alphapulse/trading/engine.py create mode 100644 web/src/apis/trading.ts create mode 100644 web/src/pages/trading/components/AddSymbolDialog.vue create mode 100644 web/src/pages/trading/index.vue diff --git a/backend/alembic/versions/20261015_a54081406f23_trading_tables.py b/backend/alembic/versions/20261015_a54081406f23_trading_tables.py new file mode 100644 index 0000000..8e77d09 --- /dev/null +++ b/backend/alembic/versions/20261015_a54081406f23_trading_tables.py @@ -0,0 +1,114 @@ +"""trading tables + +Revision ID: a54081406f23 +Revises: a7d3e9c41f02 +Create Date: 2026-10-15 01:44:51.540362 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = 'a54081406f23' +down_revision: Union[str, Sequence[str], None] = 'a7d3e9c41f02' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('trading_symbols', + sa.Column('symbol', sa.String(length=32), nullable=False), + sa.Column('angle', sa.String(length=8), nullable=False), + sa.Column('added_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('analysis_requested_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('analysis_reasons', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.PrimaryKeyConstraint('symbol') + ) + op.create_table('trading_analyses', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('symbol', sa.String(length=32), nullable=False), + sa.Column('angle', sa.String(length=8), nullable=False), + sa.Column('reasons', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('dry_run', sa.Boolean(), nullable=False), + sa.Column('status', sa.String(length=16), nullable=False), + sa.Column('action', sa.String(length=8), nullable=True), + sa.Column('decision', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('sizing', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('text', sa.Text(), nullable=True), + sa.Column('transcript', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('provider_id', sa.Integer(), nullable=True), + sa.Column('model', sa.String(length=128), nullable=True), + sa.Column('reasoning_effort', sa.String(length=8), nullable=True), + sa.Column('usage', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('error', sa.Text(), nullable=True), + sa.Column('started_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('plan_status', sa.String(length=16), nullable=True), + sa.Column('valid_until', sa.DateTime(timezone=True), nullable=True), + sa.Column('first_fill_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('close_deadline', sa.DateTime(timezone=True), nullable=True), + sa.Column('closed_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('close_reason', sa.String(length=32), nullable=True), + sa.Column('result', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.ForeignKeyConstraint(['provider_id'], ['ai_providers.id'], ondelete='SET NULL'), + sa.PrimaryKeyConstraint('id') + ) + op.create_index('ix_trading_analyses_symbol_started', 'trading_analyses', ['symbol', 'started_at'], unique=False) + op.create_table('trading_events', + sa.Column('id', sa.BigInteger(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('symbol', sa.String(length=32), nullable=True), + sa.Column('analysis_id', sa.Integer(), nullable=True), + sa.Column('kind', sa.String(length=32), nullable=False), + sa.Column('level', sa.String(length=8), nullable=False), + sa.Column('message', sa.Text(), nullable=False), + sa.Column('data', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.ForeignKeyConstraint(['analysis_id'], ['trading_analyses.id'], ), + sa.PrimaryKeyConstraint('id') + ) + op.create_index('ix_trading_events_symbol_id', 'trading_events', ['symbol', 'id'], unique=False) + op.create_table('trading_orders', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('analysis_id', sa.Integer(), nullable=True), + sa.Column('symbol', sa.String(length=32), nullable=False), + sa.Column('purpose', sa.String(length=8), nullable=False), + sa.Column('leg', sa.Integer(), nullable=True), + sa.Column('is_algo', sa.Boolean(), nullable=False), + sa.Column('side', sa.String(length=4), nullable=False), + sa.Column('type', sa.String(length=24), nullable=False), + sa.Column('price', sa.Numeric(), nullable=True), + sa.Column('trigger_price', sa.Numeric(), nullable=True), + sa.Column('quantity', sa.Numeric(), nullable=False), + sa.Column('reduce_only', sa.Boolean(), nullable=False), + sa.Column('status', sa.String(length=16), nullable=False), + sa.Column('exchange_id', sa.String(length=32), nullable=True), + sa.Column('filled_qty', sa.Numeric(), nullable=False), + sa.Column('avg_price', sa.Numeric(), nullable=True), + sa.Column('error', sa.Text(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['analysis_id'], ['trading_analyses.id'], ), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_trading_orders_analysis_id'), 'trading_orders', ['analysis_id'], unique=False) + op.create_index(op.f('ix_trading_orders_symbol'), 'trading_orders', ['symbol'], unique=False) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f('ix_trading_orders_symbol'), table_name='trading_orders') + op.drop_index(op.f('ix_trading_orders_analysis_id'), table_name='trading_orders') + op.drop_table('trading_orders') + op.drop_index('ix_trading_events_symbol_id', table_name='trading_events') + op.drop_table('trading_events') + op.drop_index('ix_trading_analyses_symbol_started', table_name='trading_analyses') + op.drop_table('trading_analyses') + op.drop_table('trading_symbols') + # ### end Alembic commands ### diff --git a/backend/src/alphapulse/api/main.py b/backend/src/alphapulse/api/main.py index 6d5d7c5..4a7db41 100644 --- a/backend/src/alphapulse/api/main.py +++ b/backend/src/alphapulse/api/main.py @@ -2,7 +2,7 @@ from fastapi import Depends, FastAPI from fastapi.openapi.docs import get_swagger_ui_html from fastapi.responses import HTMLResponse -from alphapulse.api import auth, chat, crawler, market, news, settings, skills +from alphapulse.api import auth, chat, crawler, market, news, settings, skills, trading from alphapulse.core.auth import require_admin # 关闭默认文档路由，改为登录后才能访问 @@ -15,6 +15,7 @@ app.include_router(market.ws_router) app.include_router(settings.router) app.include_router(skills.router) app.include_router(chat.router) +app.include_router(trading.router) @app.get("/api/health", include_in_schema=False) diff --git a/backend/src/alphapulse/api/trading.py b/backend/src/alphapulse/api/trading.py new file mode 100644 index 0000000..99d6a40 --- /dev/null +++ b/backend/src/alphapulse/api/trading.py @@ -0,0 +1,271 @@ +"""模拟盘页：引擎状态、启停、交易币种、时间线。账户和持仓来自 trader 进程写入 Redis 的快照""" + +import json +from datetime import datetime, timedelta +from typing import Annotated, Literal + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from pydantic import BaseModel +from sqlalchemy import func, or_, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from alphapulse.ai import llm +from alphapulse.api.crawler import LOCAL_TZ +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.trading.client import ExchangeError, public_get +from alphapulse.trading.engine import DAILY_ANALYSIS_LIMIT, OWN_CLIENT_ID, STATUS_KEY + +router = APIRouter(prefix="/api/trading", tags=["trading"], dependencies=[Depends(require_admin)]) + +Angle = Literal["short", "long"] +ANGLE_LABELS = {"short": "短线", "long": "长线"} +SYMBOLS_CACHE_KEY = f"{KEY_PREFIX}trading:symbols" +SYMBOLS_CACHE_SECONDS = 3600 +# 删除或切换角度前要求快照足够新，否则无法确认没有仓位和挂单 +SNAPSHOT_FRESH_SECONDS = 30 + + +class TodayUsage(BaseModel): + analyses: int + limit: int + input_tokens: int + output_tokens: int + + +class StatusOut(BaseModel): + configured: bool + running: bool + running_changed_at: datetime | None + provider_id: int | None + reasoning_effort: str | None + # trader 进程的心跳快照；进程离线时为空 + snapshot: dict | None + today: TodayUsage + + +class ConfigIn(BaseModel): + provider_id: int | None = None + reasoning_effort: llm.Effort | None = None + + +class SymbolOption(BaseModel): + symbol: str + base: str + + +class SymbolOut(BaseModel): + symbol: str + angle: str + added_at: datetime + analysis_requested_at: datetime | None + analyzing: bool + # 当前计划的状态：pending / open，没有计划为空 + plan_status: str | None + + +class SymbolIn(BaseModel): + symbol: str + angle: Angle + + +class SymbolPatch(BaseModel): + angle: Angle + + +class EventOut(BaseModel): + id: int + created_at: datetime + symbol: str | None + analysis_id: int | None + kind: str + level: str + message: str + data: dict | None + + +async def get_account(session: AsyncSession) -> TradingAccount: + return await session.get(TradingAccount, 1) or TradingAccount(id=1, running=False) + + +async def read_snapshot() -> dict | None: + raw = await redis.get(STATUS_KEY) + return json.loads(raw) if raw else None + + +@router.get("/status") +async def get_status(session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut: + account = await get_account(session) + today = datetime.now(LOCAL_TZ).replace(hour=0, minute=0, second=0, microsecond=0) + count, input_tokens, output_tokens = (await session.execute( + select( + func.count(), + func.coalesce(func.sum(TradingAnalysis.usage["input_tokens"].as_integer()), 0), + func.coalesce(func.sum(TradingAnalysis.usage["output_tokens"].as_integer()), 0), + ).where(TradingAnalysis.started_at >= today) + )).one() + return StatusOut( + configured=account.api_key_encrypted is not None, running=account.running, + running_changed_at=account.running_changed_at, provider_id=account.provider_id, + reasoning_effort=account.reasoning_effort, snapshot=await read_snapshot(), + today=TodayUsage(analyses=count, limit=DAILY_ANALYSIS_LIMIT, input_tokens=input_tokens, output_tokens=output_tokens), + ) + + +async def set_running(session: AsyncSession, running: bool) -> StatusOut: + account = await get_account(session) + if running and account.api_key_encrypted is None: + raise HTTPException(status.HTTP_400_BAD_REQUEST, "还没有填写模拟盘 key，请先到环境变量页填写") + if account.running != running: + account.running = running + account.running_changed_at = datetime.now(LOCAL_TZ) + session.add(account) + session.add(TradingEvent( + kind="started" if running else "stopped", + message="开始行情交易" if running else "停止行情交易：不再分析，撤销未成交的入场单；已有持仓的止损、止盈等程序规则继续执行", + )) + await session.commit() + return await get_status(session) + + +@router.post("/start") +async def start(session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut: + return await set_running(session, True) + + +@router.post("/stop") +async def stop(session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut: + return await set_running(session, False) + + +@router.patch("/config") +async def patch_config(body: ConfigIn, session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut: + account = await get_account(session) + fields = body.model_dump(exclude_unset=True) + if fields.get("provider_id") is not None and await session.get(AiProvider, fields["provider_id"]) is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "模型配置不存在") + for key, value in fields.items(): + setattr(account, key, value) + session.add(account) + await session.commit() + return await get_status(session) + + +async def symbol_options() -> list[dict]: + """可交易的币：模拟盘和实盘都在交易的 USDT 永续，只含加密货币（排除美股、商品等 TradFi）""" + if cached := await redis.get(SYMBOLS_CACHE_KEY): + return json.loads(cached) + + def crypto_perpetuals(info: dict) -> dict[str, str]: + return { + s["symbol"]: s["baseAsset"] for s in info["symbols"] + if s["contractType"] == "PERPETUAL" and s.get("underlyingType") == "COIN" + and s["status"] == "TRADING" and s["quoteAsset"] == "USDT" + } + + try: + demo = crypto_perpetuals(await public_get("/fapi/v1/exchangeInfo")) + except ExchangeError as e: + raise HTTPException(status.HTTP_502_BAD_GATEWAY, str(e)) from None + # 分析用实盘行情，执行在模拟盘，两边都要有 + mainnet = crypto_perpetuals(await call_binance(market_client.get("um", "exchange_info"))) + options = [{"symbol": s, "base": b} for s, b in sorted(demo.items()) if s in mainnet] + await redis.set(SYMBOLS_CACHE_KEY, json.dumps(options), ex=SYMBOLS_CACHE_SECONDS) + return options + + +@router.get("/symbol-options") +async def list_symbol_options() -> list[SymbolOption]: + return [SymbolOption.model_validate(o) for o in await symbol_options()] + + +@router.get("/symbols") +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()) + 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), + ) + for s in symbols + ] + + +@router.post("/symbols", status_code=status.HTTP_201_CREATED) +async def add_symbol(body: SymbolIn, session: Annotated[AsyncSession, Depends(get_session)]) -> None: + symbol = body.symbol.strip().upper() + if not any(o["symbol"] == symbol for o in await symbol_options()): + raise HTTPException(status.HTTP_400_BAD_REQUEST, f"{symbol} 不是模拟盘和实盘都在交易的加密货币 USDT 永续合约") + session.add(TradingSymbol(symbol=symbol, angle=body.angle)) + session.add(TradingEvent(symbol=symbol, kind="symbol_added", message=f"加入交易列表，角度：{ANGLE_LABELS[body.angle]}")) + try: + await session.commit() + except IntegrityError: + raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 已在交易列表中") from None + + +async def ensure_idle(session: AsyncSession, symbol: str, action: str) -> TradingSymbol: + """只有空仓、没有本程序的挂单、没有进行中的计划和分析时，才能删除或切换角度""" + s = await session.get(TradingSymbol, symbol) + if s is None: + 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)), + )) + if busy: + raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 有进行中的分析或计划，不能{action}") + snapshot = await read_snapshot() + synced_at = snapshot and snapshot["synced_at"] and datetime.fromisoformat(snapshot["synced_at"]) + if not synced_at or datetime.now(LOCAL_TZ) - synced_at > timedelta(seconds=SNAPSHOT_FRESH_SECONDS): + raise HTTPException(status.HTTP_409_CONFLICT, f"无法确认 {symbol} 的仓位和挂单（引擎离线或连接币安失败），暂时不能{action}") + if any(p["symbol"] == symbol for p in snapshot["positions"]): + raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 有持仓，不能{action}") + if any(o["symbol"] == symbol and OWN_CLIENT_ID.match(o["client_id"]) for o in snapshot["orders"] + snapshot["algo_orders"]): + raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 有本程序的挂单，不能{action}") + return s + + +@router.patch("/symbols/{symbol}") +async def patch_symbol(symbol: str, body: SymbolPatch, session: Annotated[AsyncSession, Depends(get_session)]) -> None: + s = await ensure_idle(session, symbol, "切换角度") + if s.angle != body.angle: + session.add(TradingEvent(symbol=symbol, kind="angle_changed", + message=f"角度从{ANGLE_LABELS[s.angle]}改为{ANGLE_LABELS[body.angle]}")) + s.angle = body.angle + await session.commit() + + +@router.delete("/symbols/{symbol}") +async def delete_symbol(symbol: str, session: Annotated[AsyncSession, Depends(get_session)]) -> None: + """只移出列表；分析、订单流水和时间线保留""" + s = await ensure_idle(session, symbol, "删除") + await session.delete(s) + session.add(TradingEvent(symbol=symbol, kind="symbol_removed", message="移出交易列表")) + await session.commit() + + +@router.get("/events") +async def list_events( + session: Annotated[AsyncSession, Depends(get_session)], + symbol: str | None = None, + before: int | None = None, + limit: Annotated[int, Query(ge=1, le=200)] = 50, +) -> list[EventOut]: + """按 id 倒序分页；指定 symbol 时同时返回账户级事件（启停等）""" + query = select(TradingEvent).order_by(TradingEvent.id.desc()).limit(limit) + if symbol: + query = query.where(or_(TradingEvent.symbol == symbol, TradingEvent.symbol.is_(None))) + if before: + query = query.where(TradingEvent.id < before) + return [EventOut.model_validate(e, from_attributes=True) for e in await session.scalars(query)] diff --git a/backend/src/alphapulse/models/__init__.py b/backend/src/alphapulse/models/__init__.py index 854b95b..7166efa 100644 --- a/backend/src/alphapulse/models/__init__.py +++ b/backend/src/alphapulse/models/__init__.py @@ -2,7 +2,7 @@ from alphapulse.core.db import Base from alphapulse.models.ai import AiProvider, ChatMessage, ChatSession, ContextVersion, SkillNode, SkillVersion from alphapulse.models.market import Kline, KlineSyncState, MarketSymbol from alphapulse.models.news import CrawlLog, CrawlRun, NewsItem, NewsRevision, RawBody, RawResponse, Source -from alphapulse.models.trading import TradingAccount +from alphapulse.models.trading import TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol __all__ = [ "AiProvider", @@ -23,4 +23,8 @@ __all__ = [ "SkillVersion", "Source", "TradingAccount", + "TradingAnalysis", + "TradingEvent", + "TradingOrder", + "TradingSymbol", ] diff --git a/backend/src/alphapulse/models/trading.py b/backend/src/alphapulse/models/trading.py index 68ee0d9..93a7883 100644 --- a/backend/src/alphapulse/models/trading.py +++ b/backend/src/alphapulse/models/trading.py @@ -1,6 +1,8 @@ from datetime import datetime +from decimal import Decimal -from sqlalchemy import Boolean, CheckConstraint, DateTime, ForeignKey, LargeBinary, String, func +from sqlalchemy import BigInteger, Boolean, CheckConstraint, DateTime, ForeignKey, Index, Integer, LargeBinary, Numeric, String, Text, func +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column from alphapulse.core.db import Base @@ -22,3 +24,104 @@ class TradingAccount(Base): provider_id: Mapped[int | None] = mapped_column(ForeignKey("ai_providers.id", ondelete="SET NULL")) reasoning_effort: Mapped[str | None] = mapped_column(String(8)) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class TradingSymbol(Base): + __tablename__ = "trading_symbols" + + symbol: Mapped[str] = mapped_column(String(32), primary_key=True) + # short：24h 内平仓；long：12–48h 平仓 + angle: Mapped[str] = mapped_column(String(8)) + added_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + # 待执行的分析请求：多个触发在执行前合并为一次 + analysis_requested_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + analysis_reasons: Mapped[list[str]] = mapped_column(JSONB, default=list) + + +class TradingAnalysis(Base): + """一次模型分析；action=plan 的行同时是一个交易计划，plan_* 等字段记录其生命周期""" + + __tablename__ = "trading_analyses" + __table_args__ = (Index("ix_trading_analyses_symbol_started", "symbol", "started_at"),) + + id: Mapped[int] = mapped_column(primary_key=True) + symbol: Mapped[str] = mapped_column(String(32)) + angle: Mapped[str] = mapped_column(String(8)) + reasons: Mapped[list[str]] = mapped_column(JSONB) + # 停止状态下的分析只记录计划、不下单 + dry_run: Mapped[bool] = mapped_column(Boolean) + # running / done / failed + status: Mapped[str] = mapped_column(String(16), default="running") + # plan / keep / exit + action: Mapped[str | None] = mapped_column(String(8)) + # submit_trade_plan 的参数原样保存 + decision: Mapped[dict | None] = mapped_column(JSONB) + # 程序算出的各档数量、风险金额、盈亏比等 + sizing: Mapped[dict | None] = mapped_column(JSONB) + text: Mapped[str | None] = mapped_column(Text) + transcript: Mapped[list | None] = mapped_column(JSONB) + provider_id: Mapped[int | None] = mapped_column(ForeignKey("ai_providers.id", ondelete="SET NULL")) + model: Mapped[str | None] = mapped_column(String(128)) + reasoning_effort: Mapped[str | None] = mapped_column(String(8)) + usage: Mapped[dict | None] = mapped_column(JSONB) + 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（被新计划替换） + 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)) + close_deadline: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + close_reason: Mapped[str | None] = mapped_column(String(32)) + # 平仓后的已实现盈亏、手续费 + result: Mapped[dict | None] = mapped_column(JSONB) + + +class TradingOrder(Base): + """订单流水：先写行再发单，clientOrderId / clientAlgoId 为 ap{id}，超时后按它查询而不是换 id 重发""" + + __tablename__ = "trading_orders" + + id: Mapped[int] = mapped_column(primary_key=True) + analysis_id: Mapped[int | None] = mapped_column(ForeignKey("trading_analyses.id"), index=True) + symbol: Mapped[str] = mapped_column(String(32), index=True) + # entry / tp / stop / close + purpose: Mapped[str] = mapped_column(String(8)) + # 入场或止盈的档位序号，从 0 开始 + leg: Mapped[int | None] = mapped_column(Integer) + # 条件单走 /fapi/v1/algoOrder + is_algo: Mapped[bool] = mapped_column(Boolean) + side: Mapped[str] = mapped_column(String(4)) + type: Mapped[str] = mapped_column(String(24)) + price: Mapped[Decimal | None] = mapped_column(Numeric) + trigger_price: Mapped[Decimal | None] = mapped_column(Numeric) + quantity: Mapped[Decimal] = mapped_column(Numeric) + reduce_only: Mapped[bool] = mapped_column(Boolean) + # submitting / new / partially_filled / filled / canceled / expired / rejected / finished / failed + status: Mapped[str] = mapped_column(String(16), default="submitting") + # orderId 或 algoId + exchange_id: Mapped[str | None] = mapped_column(String(32)) + filled_qty: Mapped[Decimal] = mapped_column(Numeric, default=0) + avg_price: Mapped[Decimal | None] = mapped_column(Numeric) + error: Mapped[str | None] = mapped_column(Text) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class TradingEvent(Base): + """只追加的时间线""" + + __tablename__ = "trading_events" + __table_args__ = (Index("ix_trading_events_symbol_id", "symbol", "id"),) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + # 账户级事件（启停等）为空 + symbol: Mapped[str | None] = mapped_column(String(32)) + analysis_id: Mapped[int | None] = mapped_column(ForeignKey("trading_analyses.id")) + kind: Mapped[str] = mapped_column(String(32)) + # info / warn / error + level: Mapped[str] = mapped_column(String(8), default="info") + message: Mapped[str] = mapped_column(Text) + data: Mapped[dict | None] = mapped_column(JSONB) diff --git a/backend/src/alphapulse/trading/client.py b/backend/src/alphapulse/trading/client.py index ea3ec23..715783d 100644 --- a/backend/src/alphapulse/trading/client.py +++ b/backend/src/alphapulse/trading/client.py @@ -36,6 +36,18 @@ class ExchangeError(Exception): return f"币安返回 {self.status}（{self.code}）：{self.message}" +async def public_get(path: str, params: dict[str, Any] | None = None) -> Any: + """模拟盘公开接口（exchangeInfo 等），不需要 key""" + async with httpx.AsyncClient(base_url=BASE_URL, timeout=TIMEOUT_SECONDS) as http: + try: + resp = await http.get(path, params=params) + except httpx.TransportError as e: + raise ExchangeError(None, None, f"无法连接币安模拟盘：{e!r}") from e + if resp.status_code >= 400: + raise ExchangeError(resp.status_code, None, resp.text[:200]) + return resp.json() + + class DemoClient: def __init__(self, api_key: str, api_secret: str) -> None: self.secret = api_secret.encode() diff --git a/backend/src/alphapulse/trading/engine.py b/backend/src/alphapulse/trading/engine.py new file mode 100644 index 0000000..f0b2e3f --- /dev/null +++ b/backend/src/alphapulse/trading/engine.py @@ -0,0 +1,159 @@ +"""模拟盘交易进程：定时把账户、持仓、挂单快照写入 Redis，供页面读取。 + +用法：python -m alphapulse.trading.engine +""" + +import asyncio +import json +import logging +import re +import time +from datetime import UTC, datetime + +from sqlalchemy import select + +from alphapulse.core.crypto import decrypt +from alphapulse.core.db import SessionLocal +from alphapulse.core.redis import KEY_PREFIX, redis +from alphapulse.models import TradingAccount, TradingSymbol +from alphapulse.trading.client import DemoClient, ExchangeError + +logger = logging.getLogger("alphapulse.trading.engine") + +# 进程挂掉后快照自动过期，页面据此显示「引擎离线」 +STATUS_KEY = f"{KEY_PREFIX}trader:status" +STATUS_TTL_SECONDS = 15 +SNAPSHOT_SECONDS = 3 +# 杠杆和保证金模式很少变，按币缓存 +SYMBOL_CONFIG_SECONDS = 60 +# 所有币合计每天（北京时间）最多自动分析的次数 +DAILY_ANALYSIS_LIMIT = 60 +# 本程序下的单 clientOrderId / clientAlgoId 为 ap{流水 id}；其他（含用户手动下的）不动 +OWN_CLIENT_ID = re.compile(r"^ap\d+$") + + +class Snapshot: + def __init__(self) -> None: + self.client: DemoClient | None = None + self.key_ciphertext: bytes | None = None + # 最近一次成功同步的数据；同步失败时保留并标注错误，避免页面闪空 + self.data: dict | None = None + self.synced_at: datetime | None = None + self.symbol_config: dict[str, tuple[float, dict]] = {} + + async def ensure_client(self, account: TradingAccount | None) -> DemoClient | None: + ciphertext = account.api_key_encrypted if account else None + if ciphertext != self.key_ciphertext: + # key 被修改或删除：旧连接作废，旧数据属于另一个账户，一并清掉 + if self.client is not None: + await self.client.http.aclose() + self.client = None + self.data = self.synced_at = None + self.symbol_config.clear() + 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 leverage_and_margin(self, client: DemoClient, symbol: str) -> dict: + cached = self.symbol_config.get(symbol) + if cached is None or time.monotonic() - cached[0] > SYMBOL_CONFIG_SECONDS: + config = (await client.request("GET", "/fapi/v1/symbolConfig", {"symbol": symbol}))[0] + cached = (time.monotonic(), {"leverage": config["leverage"], "margin_type": config["marginType"]}) + self.symbol_config[symbol] = cached + return cached[1] + + async def fetch(self, client: DemoClient, symbols: list[str]) -> dict: + account = await client.request("GET", "/fapi/v3/account") + positions = [p for p in await client.request("GET", "/fapi/v3/positionRisk") if float(p["positionAmt"]) != 0] + # 挂单只查交易列表里的币和有持仓的币：不带 symbol 查全部的权重是 40，3 秒一次太贵 + watched = sorted(set(symbols) | {p["symbol"] for p in positions}) + orders, algo_orders = [], [] + for symbol in watched: + orders += await client.request("GET", "/fapi/v1/openOrders", {"symbol": symbol}) + algo_orders += await client.request("GET", "/fapi/v1/openAlgoOrders", {"symbol": symbol}) + return { + "account": { + "wallet_balance": account["totalWalletBalance"], + "margin_balance": account["totalMarginBalance"], + "available_balance": account["availableBalance"], + "unrealized_pnl": account["totalUnrealizedProfit"], + }, + "positions": [ + { + "symbol": p["symbol"], + "side": "LONG" if float(p["positionAmt"]) > 0 else "SHORT", + "amount": p["positionAmt"].lstrip("-"), + "entry_price": p["entryPrice"], + "break_even_price": p["breakEvenPrice"], + "mark_price": p["markPrice"], + "unrealized_pnl": p["unRealizedProfit"], + "liquidation_price": p["liquidationPrice"], + "notional": p["notional"].lstrip("-"), + "margin": p["isolatedMargin"] if float(p["isolatedMargin"]) else p["initialMargin"], + **await self.leverage_and_margin(client, p["symbol"]), + } + for p in positions + ], + "orders": [ + { + "symbol": o["symbol"], "client_id": o["clientOrderId"], "side": o["side"], "type": o["type"], + "price": o["price"], "quantity": o["origQty"], "filled": o["executedQty"], "reduce_only": o["reduceOnly"], + "time": o["time"], + } + for o in orders + ], + "algo_orders": [ + { + "symbol": o["symbol"], "client_id": o["clientAlgoId"], "side": o["side"], "type": o["orderType"], + "trigger_price": o["triggerPrice"], "quantity": o["quantity"], "close_position": o["closePosition"], + "reduce_only": o["reduceOnly"], "time": o["createTime"], + } + for o in algo_orders + ], + } + + async def publish(self) -> None: + async with SessionLocal() as session: + account = await session.get(TradingAccount, 1) + symbols = list((await session.scalars(select(TradingSymbol.symbol))).all()) + error = None + client = await self.ensure_client(account) + if client is None: + error = "还没有填写模拟盘 key" + else: + try: + self.data = await self.fetch(client, symbols) + self.synced_at = datetime.now(UTC) + except ExchangeError as e: + logger.warning("同步模拟盘失败：%s", e) + error = str(e) + status = { + "updated_at": datetime.now(UTC).isoformat(), + "synced_at": self.synced_at.isoformat() if self.synced_at else None, + "error": error, + **(self.data or {"account": None, "positions": [], "orders": [], "algo_orders": []}), + } + await redis.set(STATUS_KEY, json.dumps(status), ex=STATUS_TTL_SECONDS) + + +async def snapshot_loop() -> None: + snapshot = Snapshot() + while True: + try: + await snapshot.publish() + except Exception: + logger.exception("发布快照失败，稍后重试") + await asyncio.sleep(SNAPSHOT_SECONDS) + + +async def main() -> None: + # 分析器和执行器在后续里程碑加入同一个 TaskGroup + async with asyncio.TaskGroup() as tg: + tg.create_task(snapshot_loop()) + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + asyncio.run(main()) diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index d7fd1d9..e7c81cd 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -24,6 +24,10 @@ services: <<: *backend command: [python, -m, alphapulse.market.worker] + trader: + <<: *backend + command: [python, -m, alphapulse.trading.engine] + web: build: context: ../web diff --git a/web/src/apis/trading.ts b/web/src/apis/trading.ts new file mode 100644 index 0000000..44966ad --- /dev/null +++ b/web/src/apis/trading.ts @@ -0,0 +1,138 @@ +import type { Effort } from "@/apis/settings" +import { http } from "@/http" + +export type Angle = "short" | "long" + +export const ANGLE_LABELS: Record = { short: "短线（24h 内平仓）", long: "长线（12–48h 平仓）" } + +/** 以下数值字段都是币安返回的十进制字符串，避免浮点误差 */ +export interface SnapshotPosition { + symbol: string + side: "LONG" | "SHORT" + amount: string + entry_price: string + break_even_price: string + mark_price: string + unrealized_pnl: string + liquidation_price: string + notional: string + margin: string + leverage: number + margin_type: string +} + +export interface SnapshotOrder { + symbol: string + client_id: string + side: "BUY" | "SELL" + type: string + price: string + quantity: string + filled: string + reduce_only: boolean + time: number +} + +export interface SnapshotAlgoOrder { + symbol: string + client_id: string + side: "BUY" | "SELL" + type: string + trigger_price: string + quantity: string + close_position: boolean + reduce_only: boolean + time: number +} + +export interface Snapshot { + updated_at: string + /** 最近一次成功同步币安的时间 */ + synced_at: string | null + error: string | null + account: { wallet_balance: string, margin_balance: string, available_balance: string, unrealized_pnl: string } | null + positions: SnapshotPosition[] + orders: SnapshotOrder[] + algo_orders: SnapshotAlgoOrder[] +} + +export interface TradingStatus { + configured: boolean + running: boolean + running_changed_at: string | null + provider_id: number | null + reasoning_effort: Effort | null + /** trader 进程离线时为空 */ + snapshot: Snapshot | null + today: { analyses: number, limit: number, input_tokens: number, output_tokens: number } +} + +export interface TradingSymbol { + symbol: string + angle: Angle + added_at: string + analysis_requested_at: string | null + analyzing: boolean + plan_status: "pending" | "open" | null +} + +export interface SymbolOption { + symbol: string + base: string +} + +export interface TradingEvent { + id: number + created_at: string + symbol: string | null + analysis_id: number | null + kind: string + level: "info" | "warn" | "error" + message: string + data: Record | null +} + +/** 本程序下的单；其他是用户手动下的 */ +export function isOwnOrder(clientId: string) { + return /^ap\d+$/.test(clientId) +} + +export function getStatus(): Promise { + return http.get("/trading/status") +} + +export function startTrading(): Promise { + return http.post("/trading/start") +} + +export function stopTrading(): Promise { + return http.post("/trading/stop") +} + +export function patchConfig(body: { provider_id?: number | null, reasoning_effort?: Effort | null }): Promise { + return http.patch("/trading/config", body) +} + +export function listSymbolOptions(): Promise { + return http.get("/trading/symbol-options") +} + +export function listSymbols(): Promise { + return http.get("/trading/symbols") +} + +export function addSymbol(symbol: string, angle: Angle): Promise { + return http.post("/trading/symbols", { symbol, angle }) +} + +export function patchSymbol(symbol: string, angle: Angle): Promise { + return http.patch(`/trading/symbols/${symbol}`, { angle }) +} + +export function deleteSymbol(symbol: string): Promise { + return http.delete(`/trading/symbols/${symbol}`) +} + +export function listEvents(params: { symbol?: string, before?: number, limit?: number }): Promise { + return http.get("/trading/events", { params }) +} diff --git a/web/src/pages/trading/components/AddSymbolDialog.vue b/web/src/pages/trading/components/AddSymbolDialog.vue new file mode 100644 index 0000000..7368cbb --- /dev/null +++ b/web/src/pages/trading/components/AddSymbolDialog.vue @@ -0,0 +1,70 @@ + + + diff --git a/web/src/pages/trading/index.vue b/web/src/pages/trading/index.vue new file mode 100644 index 0000000..4bc894e --- /dev/null +++ b/web/src/pages/trading/index.vue @@ -0,0 +1,409 @@ + + + + + diff --git a/web/src/router/index.ts b/web/src/router/index.ts index 0153bf0..86bab45 100644 --- a/web/src/router/index.ts +++ b/web/src/router/index.ts @@ -6,6 +6,7 @@ export const menuRoutes = [ { path: "/news", name: "news", component: () => import("@/pages/news/index.vue"), meta: { title: "新闻流" } }, { path: "/crawler", name: "crawler", component: () => import("@/pages/crawler/index.vue"), meta: { title: "采集控制台" } }, { path: "/market", name: "market", component: () => import("@/pages/market/index.vue"), meta: { title: "行情" } }, + { path: "/trading", name: "trading", component: () => import("@/pages/trading/index.vue"), meta: { title: "模拟盘" } }, { path: "/chat", name: "chat", component: () => import("@/pages/chat/index.vue"), meta: { title: "Chat" } }, { path: "/skills", name: "skills", component: () => import("@/pages/skills/index.vue"), meta: { title: "Skills" } }, { path: "/settings", name: "settings", component: () => import("@/pages/settings/index.vue"), meta: { title: "环境变量" } }