Python 工业级开源量化系统全流程实战

import baostock as bs
import duckdb
import pandas as pd
from datetime import datetime, timedelta

DB_PATH = "quant_data.duckdb"

def init_db(db_path: str = DB_PATH):
    """初始化 DuckDB 表结构与主键索引"""
    con = duckdb.connect(db_path)
    
    # 1. 创建前复权日线行情表
    con.execute("""
        CREATE TABLE IF NOT EXISTS daily_bars (
            symbol VARCHAR,
            trade_date DATE,
            open DOUBLE,
            high DOUBLE,
            low DOUBLE,
            close DOUBLE,
            volume BIGINT,
            amount DOUBLE,
            turnover_rate DOUBLE,
            pe_ttm DOUBLE,
            pb_mrq DOUBLE,
            PRIMARY KEY (symbol, trade_date)
        )
    """)
    
    # 2. 创建杜邦财务指标表 (含发布日期以防止前视偏差)
    con.execute("""
        CREATE TABLE IF NOT EXISTS financial_metrics (
            symbol VARCHAR,
            pub_date DATE,      -- 实际公告披露日 (严格用于回测时间对齐)
            stat_date DATE,     -- 财务报告期 (一季报/中报/三季报/年报)
            roe DOUBLE,
            net_profit_margin DOUBLE,
            asset_turnover DOUBLE,
            PRIMARY KEY (symbol, stat_date, pub_date)
        )
    """)
    con.close()
    print("DuckDB 数据库初始化完成。")

def get_latest_trade_date(con, symbol: str) -> str:
    """查询本地数据库中指定标的的最新交易日"""
    res = con.execute("SELECT MAX(trade_date) FROM daily_bars WHERE symbol = ?", [symbol]).fetchone()[0]
    if res is None:
        return "2020-01-01"  # 默认初始回溯起点
    next_date = res + timedelta(days=1)
    return next_date.strftime("%Y-%m-%d")

def fetch_and_update_daily_bars(symbols: list, db_path: str = DB_PATH):
    """增量同步全市场前复权日 K 线数据"""
    bs.login()
    con = duckdb.connect(db_path)
    today = datetime.today().strftime("%Y-%m-%d")
    fields = "date,code,open,high,low,close,volume,amount,turn,peTTM,pbMRQ"
    
    for symbol in symbols:
        start_date = get_latest_trade_date(con, symbol)
        if start_date > today:
            continue
            
        rs = bs.query_history_k_data_plus(
            symbol, fields, start_date=start_date, end_date=today,
            frequency="d", adjustflag="2"  # 2 表示前复权
        )
        
        data_list = []
        while (rs.error_code == '0') & rs.next():
            data_list.append(rs.get_row_data())
            
        if not data_list:
            continue
            
        df = pd.DataFrame(data_list, columns=rs.fields)
        df.rename(columns={
            'code': 'symbol', 'date': 'trade_date', 'turn': 'turnover_rate',
            'peTTM': 'pe_ttm', 'pbMRQ': 'pb_mrq'
        }, inplace=True)
        
        df.replace('', None, inplace=True)
        numeric_cols = ['open', 'high', 'low', 'close', 'volume', 'amount', 'turnover_rate', 'pe_ttm', 'pb_mrq']
        for col in numeric_cols:
            df[col] = pd.to_numeric(df[col], errors='coerce')
        df['trade_date'] = pd.to_datetime(df['trade_date']).dt.date

        # 使用 DuckDB 临时表批量 Upsert (插入或替换)
        con.register("df_temp", df)
        con.execute("INSERT OR REPLACE INTO daily_bars SELECT * FROM df_temp")
        con.unregister("df_temp")
        print(f"[{symbol}] 增量同步完成: {start_date} -> {today} ({len(df)} 条)")

    con.close()
    bs.logout()

import os
import duckdb
import backtrader as bt
import pandas as pd

# ================= 方案 A: DuckDB 直连 Backtrader (毫秒级注入) =================

class ExtendedPandasData(bt.feeds.PandasData):
    """扩展 Backtrader 数据源，支持 PE、PB 等多因子列"""
    lines = ('pe_ttm', 'pb_mrq', 'turnover_rate')
    params = (('pe_ttm', -1), ('pb_mrq', -1), ('turnover_rate', -1))

def load_feed_from_duckdb(db_path: str, symbol: str, start_date: str, end_date: str) -> ExtendedPandasData:
    con = duckdb.connect(db_path, read_only=True)
    query = """
        SELECT trade_date AS datetime, open, high, low, close, volume, pe_ttm, pb_mrq, turnover_rate
        FROM daily_bars WHERE symbol = ? AND trade_date BETWEEN ? AND ? ORDER BY trade_date ASC
    """
    df = con.execute(query, [symbol, start_date, end_date]).fetchdf()
    con.close()
    
    df['datetime'] = pd.to_datetime(df['datetime'])
    df.set_index('datetime', inplace=True)
    return ExtendedPandasData(dataname=df, name=symbol)

# ================= 方案 B: 批量导出并编译为 Qlib 二进制数据集 =================

def export_duckdb_to_qlib_csv(db_path: str, output_csv_dir: str):
    """利用 DuckDB 高性能分区导出为 Qlib 兼容的标准 CSV 目录"""
    os.makedirs(output_csv_dir, exist_ok=True)
    con = duckdb.connect(db_path, read_only=True)
    symbols = [row[0] for row in con.execute("SELECT DISTINCT symbol FROM daily_bars").fetchall()]
    
    print(f"开始导出 {len(symbols)} 只标的数据至 Qlib CSV 目录...")
    for sym in symbols:
        file_name = f"{sym.replace('.', '')}.csv"  # sh.600519 -> sh600519.csv
        file_path = os.path.join(output_csv_dir, file_name)
        query = f"""
            COPY (
                SELECT trade_date AS date, open, high, low, close, volume, turnover_rate, pe_ttm
                FROM daily_bars WHERE symbol = '{sym}' ORDER BY trade_date ASC
            ) TO '{file_path}' (HEADER, DELIMITER ',')
        """
        con.execute(query)
    con.close()
    print("Qlib CSV 导出完成。")

# 导出后在终端中执行编译（或在 Python 中调用 os.system）：
# python -m qlib.dump_bin --csv_path ./qlib_csv_dump --qlib_dir ~/.qlib/qlib_data/cn_data_custom --include_fields open,high,low,close,volume,turnover_rate,pe_ttm --symbol_field_name symbol --date_field_name date

import duckdb
import pandas as pd
import qlib
from qlib.constant import REG_CN
from qlib.utils import init_instance_by_config

DB_PATH = "quant_data.duckdb"

def init_prediction_table(db_path: str = DB_PATH):
    """初始化预测结果存储表"""
    con = duckdb.connect(db_path)
    con.execute("""
        CREATE TABLE IF NOT EXISTS qlib_predictions (
            model_name VARCHAR,
            trade_date DATE,
            symbol VARCHAR,
            score DOUBLE,
            score_rank BIGINT,      -- 当日横截面得分排名 (1 为最优)
            PRIMARY KEY (model_name, trade_date, symbol)
        )
    """)
    con.close()

def save_qlib_scores_to_duckdb(model, dataset, model_name: str, db_path: str = DB_PATH, segment: str = "test"):
    """执行模型推理并将结果 Upsert 写回 DuckDB"""
    pred_series = model.predict(dataset, segment=segment)
    pred_df = pred_series.to_frame(name="score") if isinstance(pred_series, pd.Series) else pred_series.copy()
    pred_df.columns = ["score"]
    
    pred_df.reset_index(inplace=True)
    pred_df.rename(columns={'datetime': 'trade_date', 'instrument': 'symbol'}, inplace=True)
    pred_df['trade_date'] = pd.to_datetime(pred_df['trade_date']).dt.date
    
    # 格式对齐: sh600519 -> sh.600519
    pred_df['symbol'] = pred_df['symbol'].apply(lambda x: f"{x[:2]}.{x[2:]}" if not '.' in x else x)
    
    # 计算当日横截面得分排名
    pred_df['score_rank'] = pred_df.groupby('trade_date')['score'].rank(ascending=False, method='min').astype(int)
    pred_df['model_name'] = model_name
    pred_df = pred_df[['model_name', 'trade_date', 'symbol', 'score', 'score_rank']]
    
    con = duckdb.connect(db_path)
    con.register("df_pred_temp", pred_df)
    con.execute("INSERT OR REPLACE INTO qlib_predictions SELECT * FROM df_pred_temp")
    con.unregister("df_pred_temp")
    con.close()
    print(f"[{model_name}] 写入完成: 共写入 {len(pred_df)} 条预测评分记录到 DuckDB。")

import backtrader as bt
import duckdb
import pandas as pd

DB_PATH = "quant_data.duckdb"

class QlibTopKRotationStrategy(bt.Strategy):
    params = (
        ('top_k', 5),                      # 目标持仓股票数量
        ('rebalance_weekday', 1),          # 每周几调仓 (1=周一; None 表示每日调仓)
        ('model_name', 'LGBM_Alpha158_v1'),# 对应 DuckDB 中的模型标签
        ('target_weight_ratio', 0.95),     # 总仓位控制 (保留 5% 现金防滑点)
        ('scores_dict', None),             # 预加载的打分哈希字典
    )

    def __init__(self):
        self.scores_dict = self.p.scores_dict or {}
        self.last_rebalance_date = None
        self.target_symbols = []

    def next(self):
        dt = self.data0.datetime.date(0)
        dt_str = dt.strftime("%Y-%m-%d")

        # 调仓周期控制 (避免同周重复触发)
        if self.p.rebalance_weekday is not None:
            if dt.weekday() != (self.p.rebalance_weekday - 1) and self.last_rebalance_date is not None:
                return
            if self.last_rebalance_date and (dt - self.last_rebalance_date).days < 4:
                return

        daily_top = self.scores_dict.get(dt_str, [])
        if not daily_top:
            return

        self.target_symbols = daily_top[:self.p.top_k]
        self.last_rebalance_date = dt
        target_weight = self.p.target_weight_ratio / len(self.target_symbols)

        # 1. 卖出已掉出 Top-K 的持仓标的
        for data in self.datas:
            symbol = data._name
            pos = self.getposition(data).size
            if pos > 0 and symbol not in self.target_symbols:
                self.order_target_percent(data, target=0.0)

        # 2. 买入或调整入选标的至目标等权
        for data in self.datas:
            symbol = data._name
            if symbol in self.target_symbols:
                self.order_target_percent(data, target=target_weight)

import math
import time
from datetime import datetime
import duckdb
from apscheduler.schedulers.blocking import BlockingScheduler
from xtquant import xtconstant, xtdata
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xttype import StockAccount

MINI_QMT_PATH = r"D:\国金证券QMT交易端\userdata_mini"  # miniQMT 本地安装路径
ACCOUNT_ID = "12345678"                             # 实盘资金账号
DB_PATH = "quant_data.duckdb"
MODEL_NAME = "LGBM_Alpha158_v1"
TOP_K = 5
TARGET_POSITION_RATIO = 0.95

class CustomTraderCallback(XtQuantTraderCallback):
    """订单与成交回报异步监听器"""
    def on_stock_order(self, order):
        print(f"[订单回报] {order.stock_code} | 状态: {order.order_status} | 报单: {order.order_volume} | 成交: {order.traded_volume}")

    def on_stock_trade(self, trade):
        print(f"[成交确认] {trade.stock_code} | 价格: {trade.traded_price:.2f} | 数量: {trade.traded_volume}")

def init_trader() -> tuple[XtQuantTrader, StockAccount]:
    """建立与本地 miniQMT 客户端的 IPC 通信通道"""
    session_id = int(time.time())
    xt_trader = XtQuantTrader(MINI_QMT_PATH, session_id)
    account = StockAccount(ACCOUNT_ID, 'STOCK')
    
    xt_trader.register_callback(CustomTraderCallback())
    xt_trader.start()
    if xt_trader.connect() != 0:
        raise ConnectionError("miniQMT 连接失败，请确保本地 QMT 客户端已处于登录状态！")
    
    xt_trader.subscribe(account)
    return xt_trader, account

def execute_daily_rebalance():
    """每日/每周定时调仓任务 (09:35 避开集合竞价剧烈跳空)"""
    today_str = datetime.today().strftime("%Y-%m-%d")
    print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 开始执行调仓任务...")

    xt_trader, account = init_trader()

    # 1. 从 DuckDB 读取当日截面 Top-K 标的
    con = duckdb.connect(DB_PATH, read_only=True)
    df_top = con.execute(
        "SELECT symbol FROM qlib_predictions WHERE model_name = ? AND trade_date = ? ORDER BY score_rank ASC LIMIT ?",
        [MODEL_NAME, today_str, TOP_K]
    ).fetchdf()
    con.close()

    if df_top.empty:
        print("未获取到当日预测信号，跳过本次调仓。")
        return

    # 代码格式转换: sh.600519 -> 600519.SH
    target_stocks = [f"{s.split('.')[1]}.{s.split('.')[0].upper()}" for s in df_top['symbol']]
    print(f"今日目标持仓 (Top-{TOP_K}): {target_stocks}")

    # 2. 查询当前资金与持仓
    asset = xt_trader.query_stock_asset(account)
    positions = xt_trader.query_stock_positions(account)
    current_pos_map = {p.stock_code: p for p in positions} if positions else {}

    # 3. 获取实时行情与涨跌停价
    all_needed_stocks = list(set(target_stocks + list(current_pos_map.keys())))
    full_ticks = xtdata.get_full_tick(all_needed_stocks)

    # ---------------- 步骤 A: 卖出调出持仓 ----------------
    for stock_code, pos in current_pos_map.items():
        if stock_code not in target_stocks and pos.can_use_volume > 0:
            tick = full_ticks.get(stock_code, {})
            if tick.get('lastPrice', 0.0) <= tick.get('downLimit', 0.0) and tick.get('downLimit', 0.0) > 0:
                print(f"[风控提示] {stock_code} 处于跌停状态，暂无法平仓。")
                continue
            
            xt_trader.order_stock_async(
                account, stock_code, xtconstant.STOCK_SELL,
                pos.can_use_volume, xtconstant.LATEST_PRICE, 0.0,
                "TopK_Rotation", "Rebalance_Sell"
            )
            print(f"发出卖单: {stock_code} | 数量: {pos.can_use_volume}")

    time.sleep(3)  # 等待卖单处理释放可用资金

    # ---------------- 步骤 B: 买入/配平目标持仓 ----------------
    asset_after = xt_trader.query_stock_asset(account)
    target_value_per_stock = (asset_after.total_asset * TARGET_POSITION_RATIO) / len(target_stocks)

    for stock_code in target_stocks:
        tick = full_ticks.get(stock_code, {})
        last_price = tick.get('lastPrice', 0.0)
        up_limit = tick.get('upLimit', 0.0)

        if last_price <= 0 or (last_price >= up_limit and up_limit > 0):
            print(f"[风控提示] {stock_code} 封死涨停或行情缺失，跳过买入。")
            continue

        # 按 100 股向下取整
        target_vol = math.floor((target_value_per_stock / last_price) / 100) * 100
        current_vol = current_pos_map[stock_code].volume if stock_code in current_pos_map else 0
        diff_vol = target_vol - current_vol

        if diff_vol >= 100:
            xt_trader.order_stock_async(
                account, stock_code, xtconstant.STOCK_BUY,
                int(diff_vol), xtconstant.LATEST_PRICE, 0.0,
                "TopK_Rotation", "Rebalance_Buy"
            )
            print(f"发出买单: {stock_code} | 增持: {diff_vol} 股")

if __name__ == "__main__":
    scheduler = BlockingScheduler(timezone="Asia/Shanghai")
    scheduler.add_job(execute_daily_rebalance, 'cron', day_of_week='mon-fri', hour=9, minute=35)
    print("实盘自动调仓引擎已就绪，等待交易日 09:35 触发...")
    scheduler.start()