Simon
ba52edc2d6
- SignalRules（买入 rank 阈值/趋势 MA/动量 + 卖出区间/破位警示）+ SignalEvent （BUY/WATCH/SELL，score/price/trigger_reason 可解释）+ SignalResult/Meta - quant/signal.generate_signals：与选股同一评分引擎取全市场 rank，按规则分类输出 - signal_snapshot/signal_event 表（migration d8e0b2f3c4d5，MySQL 已应用）+ Repo - SignalService + POST /api/signals（同步+落库）、GET 详情/列表 - tests/test_signals.py（引擎分类/排序/破位不 BUY、service、API 提交读回）；全量 pytest 通过
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
"""交易信号 API（M8.1）：提交/查询信号（落库可复现）。
|
||
|
||
POST /api/signals body: {query: SelectionQuery, rules?: SignalRules}
|
||
GET /api/signals/{id} 读回某次信号
|
||
GET /api/signals 历史信号元数据（可过滤 as_of）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from typing import Annotated
|
||
|
||
from fastapi import APIRouter, HTTPException, Query
|
||
from pydantic import BaseModel
|
||
|
||
from app.api.deps import DbSession, SignalRepoDep, SignalServiceDep
|
||
from app.application.services.job_executor import new_id
|
||
from app.domain.entities.selection import SelectionQuery
|
||
from app.domain.entities.signal import SignalMeta, SignalResult, SignalRules
|
||
|
||
router = APIRouter(prefix="/signals", tags=["signals"])
|
||
|
||
_AsOfQuery = Annotated[date | None, Query(description="按信号时点过滤")]
|
||
_LimitQuery = Annotated[int, Query(ge=1, le=200)]
|
||
|
||
|
||
class SignalRequest(BaseModel):
|
||
query: SelectionQuery
|
||
rules: SignalRules = SignalRules()
|
||
|
||
|
||
class SignalRun(BaseModel):
|
||
signal_id: str
|
||
result: SignalResult
|
||
|
||
|
||
@router.post("", response_model=SignalRun, summary="生成一次交易信号（同步）并落库")
|
||
def run_signal(
|
||
req: SignalRequest,
|
||
service: SignalServiceDep,
|
||
signal_repo: SignalRepoDep,
|
||
session: DbSession,
|
||
) -> SignalRun:
|
||
result = service.signal(req.query, req.rules)
|
||
signal_id = new_id("SIG")
|
||
signal_repo.save(signal_id, result)
|
||
session.commit()
|
||
return SignalRun(signal_id=signal_id, result=result)
|
||
|
||
|
||
@router.get("/{signal_id}", response_model=SignalResult, summary="读回一次信号结果")
|
||
def get_signal(signal_id: str, signal_repo: SignalRepoDep) -> SignalResult:
|
||
result = signal_repo.get(signal_id)
|
||
if result is None:
|
||
raise HTTPException(status_code=404, detail=f"信号记录 {signal_id} 不存在")
|
||
return result
|
||
|
||
|
||
@router.get("", response_model=list[SignalMeta], summary="历史信号元数据列表")
|
||
def list_signals(
|
||
signal_repo: SignalRepoDep,
|
||
as_of: _AsOfQuery = None,
|
||
limit: _LimitQuery = 20,
|
||
) -> list[SignalMeta]:
|
||
return signal_repo.list_recent(as_of=as_of, limit=limit)
|