前言

现在股票行情不错，大盘已经突破4000点，我在想，是不是可以根据走势、资金量和AI的一些判断是可以有大几率做出股票涨跌判断的，同时也可以自动化程序进行交易，请帮我根据这个思路写一个工程，要求可以对股票涨跌做实时判断涨跌概率，同时可以对不同股票创建不同的交易策略实现自动化交易。
现在我把我的想法结合AI做成一个可落地的工程设计与样例实现：
包含系统架构、数据流、模型设计（实时涨跌概率预测）、多策略管理、回测与纸面交易、实盘执行接口说明、风险控制与合规要点，以及关键代码片段（可直接运行或作为模板改造）。

下面内容分为：概览 → 架构设计 → 数据与特征 → 模型（实时概率预测） → 策略层与多策略管理 → 执行层（适配不同券商）→ 回测/仿真 → 监控/运维 → 关键代码样例 → 部署与注意事项。

---

一、工程概览（目标与范围）

目标：构建一个框架，能对中国 A 股（或其他市场）股票做实时涨跌概率预测，并允许为不同股票定义不同策略（动量、均值回复、波段等），同时支持回测、纸面交易与接入真实券商 API 的自动化下单（需用户接入券商、承担合规与风险责任）。

主要能力：

• 实时数据摄取（分时/分笔/盘口/成交/资金流/公告/新闻等）
• 实时特征工程与向量化
• 在线模型：输出每个短期（如 1min/5min/30min）涨跌概率
• 策略管理：每只股票可绑定不同策略与风险参数
• 执行适配器：支持多券商/仿真（stub 接口）
• 回测与仿真环境
• 风控：资金管理、逐笔限额、最大持仓/回撤/熔断
• 日志、审计与告警

风险声明（必须读）：本工程提供技术实现模板，不构成投资建议。量化交易涉及市场风险、流动性风险、执行滑点和监管合规风险。实际使用前请做严格回测、风控与合规评估。

---

二、总体架构（模块化）

+---------------------+        +--------------------+       +------------------+
| 实时数据源 (行情/资) | -----> | Data Ingest / Feat | ----> | Prediction Model |
+---------------------+        +--------------------+       +------------------+
                                                              |
                                                              v
                                                       +------------+
                                                       | Strategy   |
                                                       | Manager    |
                                                       +------------+
                                                              |
                                              +---------------+---------------+
                                              |                               |
                                       +------+-----+                   +-----+------+
                                       | Execution  |                   | Backtester |
                                       | Adapter    |                   | Simulator  |
                                       +------------+                   +------------+
                                              |
                                        Broker A / Broker B / Paper

关键组件说明：

• Data Ingest：WebSocket/REST 聚合实时tick、分钟线、资金流、新闻
• Feature Engine：在线计算技术面、统计特征、衍生指标（rolling mean, vol, order book imbalances）
• Model：在线推理服务（低延迟），输出概率分布 p(up), p(down), p(flat)
• Strategy Manager：把概率映射成信号 + 下单指令（带止损、止盈规则）
• Execution Adapter：下单、撤单、回填（支持异步回调），也支持 paper 模式
• Backtester：用历史数据做回测和参数搜索
• Monitor：日志、指标、告警、仪表盘

---

三、数据与特征（关键字段）

建议实时与历史数据源：

• 行情：逐笔成交、盘口（五档或更多）、分时、分钟K、日K（来源：券商API、行情直连或第三方如通联/聚宽/米筐/Tushare 做历史）
• 资金流：主力净流入、分笔资金大小分布
• 订单簿：买卖量不平衡、撤单比率
• 市场广度：板块涨幅、换手率、指数波动
• 衍生：rolling returns (1/3/5/15/30min), volatility, RSI, MACD, VWAP, orderflow_imbalance

典型特征示例（实时窗口）：

• last_price, last_return_1m, return_5m_std, volume_rate (当前 minuto / 过去 avg)
• bid_ask_imbalance = (sum(bidsize)-sum(asksize))/total
• recent_large_trades_count, recent_buy_pct_by_institution
• factor_score（若有基本面/财报/公告映射）

---

四、模型设计：实时涨跌概率预测

目的：给定最近 T 窗口的特征，预测短期涨跌概率（例如 1min/5min/30min）。

推荐模型架构（可扩展）：

• 基线：LightGBM / XGBoost（训练快、在线推理延迟低）
• 序列模型：1D-CNN / LSTM / Transformer（能捕捉时间序列依赖）
• 混合：先用 CNN/LSTM 提取时序特征，再用 GBDT 做最终分类回归（概率）

标签设计：

• 二分类：未来 N 分钟价格相对当前涨 > threshold（例如 +0.2%）记为 1，跌 < -0.2% 为 -1 或 0；可做多分类（涨/平/跌）
• 回归：预测未来收益分布（后续可映射成概率）

训练/在线更新：

• 离线：历史数据训练模型（cross-validation）、保存模型版本
• 在线：每天/小时增量训练或使用在线学习（如 SGD）、并做概念漂移检测

输出：

• 对每只股票和每个时间粒度，返回 p_up, p_down, expected_return, conf，以及模型版本、timestamp

延迟要求：

• 目标推理延迟 < 200ms（实时策略需要低延迟）；可用 GPU/ONNX/TF Serving 或 CPU-optimized LightGBM。

校准：

• 用 Platt scaling 或 isotonic regression 对概率进行校准（重要：概率要可解释）。

---

五、策略层与多策略系统

设计要点：

• 策略（Strategy） = 一套规则，把模型概率映射成仓位与订单（包括止损、止盈、下单类型）
• 支持多策略并行：每支股票可配置主策略与备选策略
• 策略参数化：持仓上限、最大下单量、滑点假设、最小信号阈值
• 策略生命周期：信号生成 → 委托下单 → 订单管理 → 风控干预

示例策略：

1. 动量策略（Momentum）

   • 触发条件：p_up > 0.7 且 recent_return > 0
   • 建仓：市价/限价买入，仓位 = 平均资金 * 0.02（单股）
   • 止损/止盈：-0.5% / +1.0%，或以时间止损（持有超过 60 分钟平仓）

2. 均值回复（Mean Reversion）

   • 触发条件：p_down > 0.65 且 price deviates from VWAP > 1%
   • 做空或做空等价（A 股做空受限，需用融资融券或期权）

3. 资金流驱动（Fund Flow）

   • 触发条件：短期大单买入比例 > threshold 且板块强势
   • 风险上限：当个股成交量低于阈值禁入

策略优先级、风险预算与组合层面控制由 Strategy Manager 管理。

---

六、执行层（Broker / Paper / Simulator）

执行适配器职责：

• 抽象统一接口：place_order(symbol, qty, price, side, type)、cancel_order(order_id)、get_order_status(order_id)、get_position(symbol)
• 支持同步或异步回调
• 支持回填：成交回报的匹配、滑点建模

券商接入：

• A 股实盘通常通过券商提供的 API（REST/WebSocket 或交易终端 SDK），也可使用第三方机构接口（需签约）；不同券商接口差异大，工程中留出 adapter 层实现
• 若不能接入实盘，使用 paper 模式或接入券商的仿真环境

执行注意：

• 对成交/撤单/回报做幂等处理与持久化
• 下单策略考虑限价优先、IOC、FAK 等类型以控制滑点
• 事务日志全部记录，用于审计

---

七、回测与仿真

回测模块职责：

• 离线回测历史数据（T+历史分笔与盘口最好）
• 支持回测滑点、手续费、撮合引擎（按历史盘口撮合）
• 分日回测、防止未来函数（no lookahead）
• 生成绩效指标：年化收益、夏普、最大回撤、胜率、收益分布

工具链建议：

• 自行实现轻量回测 or 使用现成库（如 zipline、backtrader、rqalpha）并扩展支持 tick/分笔数据
• 串接模型预测 pipeline 做滚动训练（walk-forward validation）

---

八、监控、日志与审计

必须实现：

• 实时指标：P&L、持仓、回撤、当日交易次数、未平仓委托
• 告警：失败下单、超限、模型异常（预测 drift）、网络断连
• 审计日志：每次信号、每次下单、每次回报必须可回溯
• Dashboard：Grafana / Kibana 展示关键运营指标

---

九、合规与风控要点（强烈建议）

• A 股自动交易必须符合中国监管与券商要求（有些券商不允许完全无人操盘）
• 需要 KYC、合规审批、资管资格（若代客）等
• 速度上要考虑交易所限速、券商限流、日内涨跌停规则
• 策略应设置全局熔断（若短期收益异常或连续失败触发暂停）

---

十、关键代码样例（Python）

下面给出简化可运行的代码片段（核心模块），用于演示概念并可作为工程模板。这是示例代码，不直接对接任何券商 API，实盘请替换 ExecutionAdapter 的实现并做充分测试。

代码结构（示例）：

realtime/
  data_collector.py
  features.py
  model_service.py
  strategy_manager.py
  execution_adapter.py
  backtester.py
  run_realtime.py

1) data_collector.py（示例：仿真/订阅行情）

# data_collector.py
import time
import threading
import queue
import random
from collections import deque

# 简易行情生成器（实际应替换为券商/行情源 WebSocket）
class MockMarketFeed:
    def __init__(self, symbols):
        self.symbols = symbols
        self.listeners = []
        self.running = False

    def start(self):
        self.running = True
        t = threading.Thread(target=self._run)
        t.daemon = True
        t.start()

    def register(self, callback):
        self.listeners.append(callback)

    def _run(self):
        while self.running:
            for s in self.symbols:
                tick = {
                    "symbol": s,
                    "ts": int(time.time()*1000),
                    "price": round(10 + random.random()*2, 3),
                    "volume": random.randint(100, 2000)
                }
                for cb in self.listeners:
                    cb(tick)
            time.sleep(0.5)

2) features.py（在线特征窗口）

# features.py
from collections import defaultdict, deque
import numpy as np
import time

class FeatureEngine:
    def __init__(self, window_seconds=300):
        self.window = window_seconds
        self.buckets = defaultdict(lambda: deque())

    def add_tick(self, tick):
        s = tick["symbol"]
        self.buckets[s].append(tick)
        now = time.time()
        # pop old
        while self.buckets[s] and (now - self.buckets[s][0]["ts"]/1000.0) > self.window:
            self.buckets[s].popleft()

    def get_features(self, symbol):
        q = list(self.buckets[symbol])
        if not q:
            return None
        prices = np.array([t["price"] for t in q])
        vols = np.array([t["volume"] for t in q])
        feat = {
            "last_price": float(prices[-1]),
            "ret_1": float((prices[-1]-prices[-2])/prices[-2]) if len(prices)>1 else 0.0,
            "vol_mean": float(np.mean(vols)),
            "vol_std": float(np.std(vols))
        }
        return feat

3) model_service.py（示例：轻量概率模型，使用历史训练好的 sklearn 模型）

# model_service.py
import numpy as np
# 假设我们有一个已训练的 sklearn model saved as joblib
# 这里演示简单阈值模型作为占位
class ModelService:
    def __init__(self):
        pass

    def predict_proba(self, features):
        # features: dict -> 返回 p_up, p_down
        # 简单启发式：若 ret_1 > 0 => p_up higher
        r = features.get("ret_1", 0)
        p_up = 0.5 + min(0.25, r*10)
        p_up = max(0.01, min(0.99, p_up))
        return {"p_up": p_up, "p_down": 1-p_up}

4) strategy_manager.py（示例策略）

# strategy_manager.py
class StrategyManager:
    def __init__(self, exec_adapter):
        self.exec = exec_adapter
        self.strategy_params = {}  # per-symbol params

    def on_tick(self, symbol, features, proba):
        # 例：简单动量策略
        params = self.strategy_params.get(symbol, {"threshold":0.7, "size":100})
        if proba["p_up"] >= params["threshold"]:
            # place buy order
            self.exec.place_order(symbol, params["size"], side="BUY")
        # 可以扩展止损/止盈和仓位管理

5) execution_adapter.py（抽象 + paper 模式）

# execution_adapter.py
import logging, time
class ExecutionAdapter:
    def __init__(self, mode="paper"):
        self.mode = mode
        self.order_id_counter = 0

    def place_order(self, symbol, qty, side, price=None, order_type="MARKET"):
        self.order_id_counter += 1
        oid = f"ORD{self.order_id_counter}"
        logging.info(f"[{self.mode}] Placing {order_type} {side} {qty} {symbol} => oid {oid}")
        # paper: simulate immediate filled
        if self.mode == "paper":
            return {"order_id": oid, "status": "FILLED", "filled_qty": qty, "price": price or 0.0, "ts": int(time.time()*1000)}
        # real broker: call broker api and return order handle
        return {"order_id": oid, "status":"SUBMITTED"}

6) run_realtime.py（主流程）

# run_realtime.py
from data_collector import MockMarketFeed
from features import FeatureEngine
from model_service import ModelService
from strategy_manager import StrategyManager
from execution_adapter import ExecutionAdapter

symbols = ["600000.SH", "000001.SZ"]
feed = MockMarketFeed(symbols)
feat = FeatureEngine()
model = ModelService()
exec_adp = ExecutionAdapter(mode="paper")
strat = StrategyManager(exec_adp)

def on_tick(tick):
    feat.add_tick(tick)
    f = feat.get_features(tick["symbol"])
    if not f:
        return
    proba = model.predict_proba(f)
    strat.on_tick(tick["symbol"], f, proba)

feed.register(on_tick)
feed.start()

# keep alive
import time
while True:
    time.sleep(1)

上面代码是最小可运行演示（paper 模式）。实际项目需替换 model_service 为训练好的模型，data_collector 通过券商/行情源 WebSocket 获取真实数据，execution_adapter 对接券商下单 API。

---

十一、回测示例（思路）

• 使用历史分钟或 tick 数据，按时间序列推进：

  • 在每个时间步用历史切片生成特征 → 模型预测（注意不要用未来信息）→ 生成信号 → 以假设撮合规则撮合成交（按历史盘口）

• 计算手续费、滑点、可用资金约束

• Use walk-forward CV 做参数稳健性测试

---

十二、部署、运维与性能考虑

• 推理服务（ModelService）建议打包成独立微服务，用 FastAPI/Flask 并使用 gunicorn/uvicorn。
• 用 Redis/Kafka 做消息总线（行情→特征→策略→执行）。
• 模型版本控制（MLflow 或 DVC），并记录特征分布以便 drift 检测。
• 高可用：主从或多副本部署，状态持久化到 DB（orders、positions）。
• 安全：对接券商时用 TLS、API key 存在 Secrets Manager。

---

十三、合规与实际接入券商的说明（重要）

• 中国 A 股实盘自动化交易需与券商签约，了解券商 API 的速率限制、下单品种（A 股做空受限）、T+1 交割、融资融券规则等。
• 许多券商对自动化下单有审核与测试环境，接入前需与券商商务/合规沟通。
• 若要为客户提供服务，需遵守当地资管/基金/经纪监管要求。

---

以下是Github

说明与免责声明（重要）

• 本仓库为技术样板，仅用于研究、回测与纸面交易演示。
• 不包含与任何券商的实盘凭证或自动下单权限；若要接入实盘，需你与券商签约并在 ExecutionAdapter 中实现券商 API 调用（我在文档内给出模版与接入建议）。
• 量化和自动化交易有高风险，务必做充分回测与风控、并遵守监管与券商规则。

---

项目总体（名字）

项目名：a-shares-agentic-trader（示例）

目录结构（主要文件）

a-shares-agentic-trader/
├─ docker/
│  └─ Dockerfile
├─ realtime/
│  ├─ data_collector.py
│  ├─ features.py
│  ├─ model_service.py
│  ├─ strategy_manager.py
│  ├─ execution_adapter.py
│  ├─ run_realtime.py
│  └─ requirements.txt
├─ training/
│  ├─ make_dataset.py
│  ├─ train_lgb.py
│  ├─ infer.py
│  └─ requirements.txt
├─ backtest/
│  ├─ backtester.py
│  ├─ walk_forward.py
│  └─ requirements.txt
├─ docker-compose.yml
├─ README.md
└─ LICENSE

我会把每个文件的核心实现/内容贴出来（注：为避免单条回复过长，我在关键地方保留合理注释；如需我把完整仓库打包成一个可下载的压缩文件或 GitHub 仓库内容，我可以继续生成）。

---

一：Realtime 最小可运行系统（paper 模式）

路径：realtime/

realtime/requirements.txt

numpy
pandas
scikit-learn
lightgbm
flask
requests
joblib

realtime/data_collector.py

最小行情源（demo 模拟或接 WebSocket）。

# realtime/data_collector.py
import time, threading, random
from typing import Callable

class MockMarketFeed:
    """模拟行情：每 0.5s 对每个 symbol 推送一个 tick"""
    def __init__(self, symbols):
        self.symbols = symbols
        self._callbacks = []
        self._running = False

    def register(self, cb: Callable):
        self._callbacks.append(cb)

    def start(self):
        self._running = True
        t = threading.Thread(target=self._run, daemon=True)
        t.start()

    def _run(self):
        while self._running:
            ts = int(time.time()*1000)
            for s in self.symbols:
                tick = {"symbol": s, "ts": ts, "price": round(10+random.random()*5,3), "volume": random.randint(100,5000)}
                for cb in self._callbacks:
                    try:
                        cb(tick)
                    except Exception as e:
                        print("callback err", e)
            time.sleep(0.5)

    def stop(self):
        self._running = False

若接真实行情：把 MockMarketFeed 替换为 WebSocket 客户端（券商/第三方行情）并解析 tick。

realtime/features.py

滑窗在线特征计算。

# realtime/features.py
from collections import defaultdict, deque
import numpy as np
import time

class FeatureEngine:
    """保存最近 T 秒/条的 tick，计算简单统计特征"""
    def __init__(self, window_seconds=300):
        self.window = window_seconds
        self.buckets = defaultdict(lambda: deque())

    def add_tick(self, tick):
        s = tick["symbol"]
        self.buckets[s].append(tick)
        now = time.time()
        # 弹出老数据
        while self.buckets[s] and (now - self.buckets[s][0]["ts"]/1000.0) > self.window:
            self.buckets[s].popleft()

    def get_features(self, symbol):
        q = list(self.buckets[symbol])
        if not q or len(q) < 2:
            return None
        prices = np.array([t["price"] for t in q])
        vols = np.array([t["volume"] for t in q])
        feat = {
            "last_price": float(prices[-1]),
            "ret_1": float((prices[-1]-prices[-2])/prices[-2]) if len(prices)>1 and prices[-2]!=0 else 0.0,
            "ret_5": float((prices[-1]-prices[0])/prices[0]) if prices[0]!=0 else 0.0,
            "vol_mean": float(np.mean(vols)),
            "vol_std": float(np.std(vols))
        }
        return feat

realtime/model_service.py

加载 LightGBM 或提供简单启发式 fallback。

# realtime/model_service.py
import joblib, os
import numpy as np

class ModelService:
    def __init__(self, model_path=None):
        self.model = None
        if model_path and os.path.exists(model_path):
            self.model = joblib.load(model_path)
        else:
            print("[ModelService] No model found, using heuristic")

    def predict_proba(self, feat: dict):
        if self.model:
            # 期望模型接收与训练时一致的 feature vector 顺序
            X = np.array([[feat.get("last_price",0), feat.get("ret_1",0), feat.get("ret_5",0), feat.get("vol_mean",0), feat.get("vol_std",0)]])
            p = self.model.predict_proba(X)[0]
            return {"p_up": float(p[1]), "p_down": float(p[0])}
        else:
            # 简单启发式
            r = feat.get("ret_1",0)
            p_up = max(0.01, min(0.99, 0.5 + r*10))
            return {"p_up": p_up, "p_down": 1-p_up}

realtime/execution_adapter.py

抽象 + paper 模式实现 + 实盘模版（钩子）。

# realtime/execution_adapter.py
import time, logging

class ExecutionAdapter:
    def __init__(self, mode="paper"):
        self.mode = mode
        self._oid = 0
        logging.basicConfig(level=logging.INFO)

    def _next_oid(self):
        self._oid += 1
        return f"ORD{int(time.time())}{self._oid}"

    def place_order(self, symbol, qty, side, price=None, order_type="MARKET"):
        oid = self._next_oid()
        logging.info(f"[{self.mode}] place_order {oid} {side} {symbol} {qty} {order_type} {price}")
        if self.mode == "paper":
            return {"order_id": oid, "status": "FILLED", "filled_qty": qty, "price": price or 0.0}
        else:
            # TODO: 在此处实现与券商的接入
            # Example pseudo:
            # resp = broker_client.place_order(...)
            # return resp
            raise NotImplementedError("Broker adapter not implemented")

    def cancel_order(self, order_id):
        logging.info(f"[{self.mode}] cancel {order_id}")
        return {"order_id": order_id, "status": "CANCELLED"}

在 else 部分放置你具体券商 SDK 的调用代码（参见下面 ExecutionAdapter 接入指南）。

realtime/strategy_manager.py

把概率映射为信号的简单管理器（可扩展为多策略、多股票配置）。

# realtime/strategy_manager.py
class StrategyManager:
    def __init__(self, exec_adapter, strategy_config=None):
        self.exec = exec_adapter
        self.config = strategy_config or {}
        # example: {"600000.SH": {"type":"momentum", "threshold":0.7, "size":100}}

    def on_tick(self, symbol, feat, proba):
        cfg = self.config.get(symbol, {"type":"momentum", "threshold":0.75, "size":100})
        if cfg["type"] == "momentum":
            if proba["p_up"] >= cfg["threshold"]:
                # place buy (paper)
                self.exec.place_order(symbol, cfg["size"], side="BUY")

realtime/run_realtime.py

主流程：启动模拟 feed、特征、模型、策略。

# realtime/run_realtime.py
from data_collector import MockMarketFeed
from features import FeatureEngine
from model_service import ModelService
from execution_adapter import ExecutionAdapter
from strategy_manager import StrategyManager
import time, os

MODEL_PATH = os.environ.get("MODEL_PATH", "../training/lgb_model.joblib")

symbols = ["600000.SH", "000001.SZ"]
feed = MockMarketFeed(symbols)
feat = FeatureEngine(window_seconds=300)
model = ModelService(model_path=MODEL_PATH)
exec_adp = ExecutionAdapter(mode="paper")
strat = StrategyManager(exec_adp, strategy_config={
    "600000.SH": {"type":"momentum", "threshold":0.7, "size":100},
    "000001.SZ": {"type":"momentum", "threshold":0.72, "size":200}
})

def on_tick(tick):
    feat.add_tick(tick)
    f = feat.get_features(tick["symbol"])
    if not f:
        return
    p = model.predict_proba(f)
    print(f"[{tick['symbol']}] price {f['last_price']} p_up {p['p_up']:.3f}")
    strat.on_tick(tick['symbol'], f, p)

if __name__ == "__main__":
    feed.register(on_tick)
    feed.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        feed.stop()

---

二：训练与推理（LightGBM Pipeline）

路径：training/

training/requirements.txt

numpy
pandas
scikit-learn
lightgbm
joblib

training/make_dataset.py

生成示例训练数据或把真实 CSV 转换成特征窗口（示例使用随机数据/仿真数据）。

# training/make_dataset.py
import numpy as np, pandas as pd
import argparse, os
def gen_synthetic(symbol="SYN", n=10000):
    ts = pd.date_range("2020-01-01", periods=n, freq="T")
    price = 10 + np.cumsum(np.random.randn(n)*0.02)
    vol = np.random.randint(100,1000,n)
    df = pd.DataFrame({"ts":ts, "price":price, "vol":vol})
    return df

def featurize(df, lookback=5):
    rows=[]
    for i in range(lookback, len(df)-5):
        window = df.iloc[i-lookback:i]
        future = df.iloc[i+1:i+6]  # next 5 min
        last = df.iloc[i]
        feat = {
            "last_price": last["price"],
            "ret_1": (last["price"]-window.iloc[-2]["price"])/window.iloc[-2]["price"] if len(window)>1 else 0,
            "ret_5": (last["price"]-window.iloc[0]["price"])/window.iloc[0]["price"],
            "vol_mean": window["vol"].mean(),
            "vol_std": window["vol"].std()
        }
        # label: whether future 5-min average > threshold (e.g., 0.2%)
        fut_ret = (future["price"].mean() - last["price"]) / last["price"]
        label = 1 if fut_ret > 0.002 else 0
        feat["label"] = label
        rows.append(feat)
    return pd.DataFrame(rows)

if __name__=="__main__":
    df = gen_synthetic(n=20000)
    ds = featurize(df)
    os.makedirs("out", exist_ok=True)
    ds.to_csv("out/dataset.csv", index=False)
    print("dataset saved to out/dataset.csv")

training/train_lgb.py

训练 LightGBM 并保存模型。

# training/train_lgb.py
import pandas as pd, lightgbm as lgb, joblib
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, accuracy_score
import os

df = pd.read_csv("out/dataset.csv")
X = df[["last_price","ret_1","ret_5","vol_mean","vol_std"]]
y = df["label"]

X_train, X_val, y_train, y_val = train_test_split(X,y,test_size=0.2,random_state=42,shuffle=False)
train_data = lgb.Dataset(X_train, label=y_train)
valid_data = lgb.Dataset(X_val, label=y_val)

params = {
    "objective":"binary",
    "metric":"auc",
    "verbosity": -1,
    "boosting_type":"gbdt",
    "num_leaves":31,
    "learning_rate":0.05
}
bst = lgb.train(params, train_data, valid_sets=[valid_data], num_boost_round=200, early_stopping_rounds=20)
os.makedirs("models", exist_ok=True)
joblib.dump(bst, "models/lgb_model.joblib")
print("model saved to models/lgb_model.joblib")
# evaluate
pred = bst.predict(X_val)
print("AUC:", roc_auc_score(y_val, pred))

training/infer.py

本地推理示例（可被 realtime 的 ModelService 使用）。

# training/infer.py
import joblib, pandas as pd
bst = joblib.load("models/lgb_model.joblib")
df = pd.read_csv("out/dataset.csv").head(5)
X = df[["last_price","ret_1","ret_5","vol_mean","vol_std"]]
print(bst.predict(X))

说明：训练脚本是最小示例；真实生产需接入历史 tick/min 数据并做特征工程、交叉验证、样本权重、类别不平衡处理、概率校准（Platt / isotonic）、概念漂移检测等。

---

三：ExecutionAdapter 实盘接入模版与指南

文件已在 realtime/execution_adapter.py 给出抽象。下面是接入步骤与示例说明（放在 README 里也方便）。

接入模版（伪代码）

# 在 ExecutionAdapter 中实现 broker 下单逻辑
class BrokerAdapter(ExecutionAdapter):
    def __init__(self, api_key, secret, base_url):
        super().__init__(mode="live")
        self.api_key = api_key
        self.secret = secret
        self.base_url = base_url
        # 初始化券商 SDK 客户端，或设置 REST headers

    def place_order(self, symbol, qty, side, price=None, order_type="MARKET"):
        # 构建请求体：依券商 api 文档
        payload = {...}
        resp = requests.post(self.base_url + "/order", json=payload, headers=headers, timeout=10)
        # 解析 resp，返回统一格式 {'order_id':..., 'status':..., 'filled_qty':..., 'price':...}
        return parsed_resp

接入注意事项

1. 速率限制：券商通常限制每分钟/每秒请求数；请实现请求节流与重试。
2. 幂等性：每次下单使用唯一 client_order_id，防止重试重复下单。
3. 回报处理：接收券商的成交回报（webhook 或轮询）并更新本地订单状态。
4. 资金/持仓管理：使用券商提供的持仓接口确认可用资金后下单。
5. 安全：API key/secret 放在 Secrets Manager（K8s secret、AWS Secrets Manager），不要写在代码或仓库。
6. 测试：先在券商的 sandbox/仿真环境完成测试。

如果你愿意，把你使用的具体券商名称或其 API 文档发给我，我可以基于其 API 写出 BrokerAdapter 的具体实现代码（并处理鉴权、签名、下单/撤单/查询接口）。

---

四：Backtester 与 Walk-Forward 实现

路径：backtest/

backtest/requirements.txt

pandas
numpy
matplotlib
joblib

backtest/backtester.py

最简回测器：滚动时间序列，调用模型、策略并按简单撮合规则计算 P&L。

# backtest/backtester.py
import pandas as pd, numpy as np
from collections import deque

class SimpleBacktester:
    def __init__(self, price_series, model, strategy, cash=100000):
        self.prices = price_series  # pd.Series indexed by datetime
        self.model = model
        self.strategy = strategy
        self.cash = cash
        self.position = 0
        self.trades = []

    def run(self):
        window = deque()
        for idx, price in self.prices.items():
            # build feature from recent window (this is simplistic)
            window.append(price)
            if len(window) > 10: window.popleft()
            if len(window) < 5: continue
            feat = {"last_price": price, "ret_1": (price-window[-2])/window[-2], "ret_5": (price-window[0])/window[0], "vol_mean":1, "vol_std":1}
            proba = self.model.predict_proba(feat)
            action = self.strategy.decide(proba)
            if action=="BUY" and self.cash>price:
                qty = int(self.cash/(10*price))  # risk sizing simple
                if qty>0:
                    self.cash -= qty*price
                    self.position += qty
                    self.trades.append(("BUY", idx, price, qty))
            if action=="SELL" and self.position>0:
                qty = self.position
                self.cash += qty*price
                self.trades.append(("SELL", idx, price, qty))
                self.position = 0
        # compute final pnl
        final_price = list(self.prices)[-1]
        nav = self.cash + self.position * final_price
        return {"nav": nav, "trades": self.trades}

backtest/walk_forward.py

简单的 Walk-Forward：按时间分块做训练与测试。

# backtest/walk_forward.py
import pandas as pd, joblib, os
from training.make_dataset import gen_synthetic, featurize
from training.train_lgb import train_lgb # or call subprocess
from backtester import SimpleBacktester

# 伪代码：分段训练+测试
def walk_forward_demo():
    df = gen_synthetic(n=10000)
    ds = featurize(df)
    # split into windows
    n = len(ds)
    step = int(n*0.2)
    results = []
    for start in range(0, n-step, step):
        train = ds.iloc[:start+step]
        test = ds.iloc[start+step:start+2*step]
        # train model (call external training script) -> get model
        # load model
        # run backtester on test
        # save metrics
    print("walk-forward demo done")

说明：以上是最小化的回测与 walk-forward 框架。生产级回测请用 tick/盘口数据、撮合引擎、手续费与滑点模型，以及日内逐笔撮合，避免未来函数。

---

五：Docker / docker-compose（可选）

docker/Dockerfile

FROM python:3.10-slim
WORKDIR /app
COPY realtime/requirements.txt /app/
RUN pip install --no-cache-dir -r requirements.txt
COPY realtime /app
CMD ["python","run_realtime.py"]

docker-compose.yml（示例）

version: "3.8"
services:
  realtime:
    build: ./docker
    environment:
      - MODEL_PATH=/app/models/lgb_model.joblib
    volumes:
      - ./training/models:/app/models
    restart: unless-stopped

---

六：README（简要运行说明）

把下面内容写入 README.md，也粘在这里方便复制。

README.md（要点）

1. 安装并准备 Python：python3 -m venv venv && source venv/bin/activate
2. 训练模型（示例数据）：

cd training
pip install -r requirements.txt
python make_dataset.py        # 生成 out/dataset.csv
python train_lgb.py           # 输出 models/lgb_model.joblib

3. 启动 realtime（paper 模式）：

cd realtime
pip install -r requirements.txt
export MODEL_PATH=../training/models/lgb_model.joblib
python run_realtime.py

4. 回测：

cd backtest
pip install -r requirements.txt
python backtester.py  # or use walk_forward.py

5. 接入实盘：修改 realtime/execution_adapter.py 中的 broker 部分，使用券商提供的 SDK/REST API，并在安全环境中存放 API keys。

---

七：风控、监控与运维建议（简短）

• 强制止损与总仓位上限（例如单股最大市值占比、日内最大交易次数）。
• 全量日志、下单/回报审计、异常告警（用 Prometheus + Grafana / ELK）。
• 模型监控：特征分布 drift、预测分布 drift、在线 A/B 测试模型版本。
• 高可用：分离数据采集、特征、推理、策略与执行微服务；使用消息队列（Kafka/Redis Streams）连接链路。
• secrets 管理：K8s secret / Vault / AWS Secrets Manager。