AI + 量化开发工程师学习笔记 Day 33:组合构建——等权组合与目标持仓管理

第二阶段：AI Quant Model（Day21～Day35）

预计学习时间：60～90 分钟。

今天严格按照我们原定的课程计划进行：

Day32，我们已经让 Linear、LightGBM、CatBoost 三个模型分别生成 Top-K 股票信号。

今天要解决的是下一步：

选出了 5 只股票之后，如何把它们变成一个真正有资金约束、有目标权重、有持仓状态的投资组合？

我们将正式区分三个概念：

• 选股信号：模型想买哪些股票。
• 目标组合：每只股票应该占总资产的多少。
• 实际持仓：考虑现金、价格、整手数量等限制后，账户真正持有什么。

今天不引入新的预测模型，也不提前实现 Day34 的完整历史回测。

一、为什么选股不等于组合构建？

假设 Day32 的 LightGBM 选出了五只股票：

我们有 10 万元资金。

应该怎样分配？

最简单的方案是等权：

当 ：

每只股票目标资金：

但问题随之出现：

如果一只股票价格为 37.60 元，分配 20000 元，能买多少股？

A 股普通股票买入通常需要满足 100 股整数倍的规则（具体证券品种及交易规则可能不同）。

因此，实际可以买入：

实际使用资金：

剩余资金：

所以：

目标权重 20%，不代表最终实际持仓权重一定恰好 20%。

二、今天必须掌握的五个概念

2.1 Target Weight：目标权重

模型希望组合达到的权重。

例如：

A 20%
B 20%
C 20%
D 20%
E 20%

2.2 Actual Weight：实际权重

根据当前市场价格和实际持仓计算：

其中：

• ：实际持股数量。
• ：当前估值价格。
• ：账户总资产净值。

2.3 Cash：现金

这里先不考虑负债、融资、费用应计等复杂账户项目。

2.4 Drift：权重偏离

如果：

目标权重：20%
实际权重：16%

则：

意味着需要考虑增持。

2.5 Rebalance：再平衡

再平衡是把当前持仓向目标组合调整。

例如：

当前：
A 30%
B 20%
C 20%
D 10%
现金 20%

目标：
A 20%
B 20%
C 20%
D 20%
E 20%

那么需要减少 A、增加 D 和 E，并使用现金。

注意：真实交易必须根据订单成交结果更新持仓，不能直接把目标权重当作已经成交的持仓。

三、组合构建的数学模型

3.1 等权组合

假设选股集合：

目标权重：

满足：

且：

这是一个不使用杠杆、不做空的满仓股票组合。

但由于交易整手限制，实际持仓通常会留有少量现金。

3.2 组合净值

组合总资产：

例如：

这才是我们今天要构建的组合数据结构。

四、Day33 工程设计

今天新增以下模块：

quant_ml/
├── src/
│   └── portfolio/
│       ├── __init__.py
│       ├── topk_selector.py        # Day32已有
│       └── equal_weight.py         # Day33新增
│
├── scripts/
│   └── build_equal_weight.py       # Day33新增
│
├── tests/
│   └── test_equal_weight.py        # Day33新增
│
└── data/
    └── portfolios/
        ├── lightgbm_target.parquet
        ├── lightgbm_orders_demo.csv
        └── lightgbm_account_demo.csv

我们明确区分：

今天实现目标组合、账户快照和订单意图。

不会把订单意图直接视为真实成交。

五、【正式工程代码】src/portfolio/equal_weight.py

这是今天的核心文件。

我们实现三个函数：

1. build_equal_weight_target()：构建目标组合。
2. calculate_account_snapshot()：计算当前净值与实际权重。
3. generate_order_intents()：根据目标权重生成计划交易数量。

完整代码如下：

# ============================================================
# Day33：等权组合构建
# - 目标权重（Target Portfolio）
# - 账户快照（NAV / Actual Weight）
# - 订单意图（Order Intent，不是成交）
# ============================================================

from __future__ import annotations

from dataclasses import dataclass
import math

import numpy as np
import pandas as pd


@dataclass(frozen=True)
classAccount:
"""
    账户当前状态。

    cash:
        可用现金，单位元。

    positions:
        股票代码 -> 当前持股数量。
    """
    cash: float
    positions: dict[str, int]


defbuild_equal_weight_target(
    signals: pd.DataFrame,
) -> pd.DataFrame:
"""
    将 Day32 的 Top-K 信号转换为等权目标组合。

    输入：
        signal_date
        symbol
        prediction
        rank

    输出：
        signal_date
        symbol
        target_weight
    """

    required = [
"signal_date",
"symbol",
    ]

    missing = [
        col for col in required
if col notin signals.columns
    ]

if missing:
raise ValueError(
f"缺少字段：{missing}"
        )

    df = signals.copy()

if df.empty:
raise ValueError(
"选股信号不能为空"
        )

    df["signal_date"] = pd.to_datetime(
        df["signal_date"]
    )

if df[
        ["signal_date", "symbol"]
    ].isna().any().any():
raise ValueError(
"信号日期或股票代码不能为空"
        )

    df["symbol"] = df[
"symbol"
    ].astype(str)

if df.duplicated(
        ["signal_date", "symbol"]
    ).any():
raise ValueError(
"存在重复股票信号"
        )

    counts = df.groupby(
"signal_date"
    )["symbol"].transform("size")

    df["target_weight"] = (
1.0 / counts
    )

return df[
        [
"signal_date",
"symbol",
"target_weight",
        ]
    ].sort_values(
        ["signal_date", "symbol"]
    ).reset_index(drop=True)


defvalidate_account(
    account: Account,
):
ifnot np.isfinite(account.cash):
raise ValueError(
"现金必须为有限数值"
        )

if account.cash < 0:
raise ValueError(
"现金不能为负"
        )

for symbol, quantity in (
        account.positions.items()
    ):
if (
            isinstance(quantity, bool)
ornot isinstance(quantity, int)
or quantity < 0
        ):
raise ValueError(
f"{symbol}持仓数量不合法"
            )


defcalculate_account_snapshot(
    account: Account,
    prices: dict[str, float],
) -> tuple[float, pd.DataFrame]:
"""
    按给定估值价格计算账户净值和持仓权重。
    """

    validate_account(account)

    rows = []
    market_value = 0.0

for symbol, quantity in (
        account.positions.items()
    ):

if quantity == 0:
continue

if symbol notin prices:
raise ValueError(
f"缺少{symbol}的估值价格"
            )

        price = float(prices[symbol])

ifnot np.isfinite(price) or price <= 0:
raise ValueError(
f"{symbol}价格不合法"
            )

        value = quantity * price

        market_value += value

        rows.append({
"symbol": symbol,
"quantity": quantity,
"price": price,
"market_value": value,
        })

    nav = account.cash + market_value

if nav <= 0:
raise ValueError(
"账户净值必须大于0"
        )

    snapshot = pd.DataFrame(
        rows,
        columns=[
"symbol",
"quantity",
"price",
"market_value",
        ],
    )

    snapshot["actual_weight"] = (
        snapshot["market_value"] / nav
    )

return nav, snapshot


defgenerate_order_intents(
    target: pd.DataFrame,
    account: Account,
    prices: dict[str, float],
    lot_size: int = 100,
) -> pd.DataFrame:
"""
    生成计划交易数量。

    注意：
    - 这是订单意图，不是成交回报。
    - 卖出可以按实际持仓数量计算，
      不强制要求100股整数倍。
    - 买入按lot_size向下取整。
    - 不处理手续费、滑点、T+1可卖数量、
      涨跌停、停牌或订单成交。
    """

if lot_size <= 0:
raise ValueError(
"lot_size必须大于0"
        )

if target.empty:
raise ValueError(
"目标组合不能为空"
        )

if target["symbol"].duplicated().any():
raise ValueError(
"单期目标组合存在重复股票"
        )

    weights = pd.to_numeric(
        target["target_weight"],
        errors="raise",
    )

if (
not np.isfinite(weights).all()
or (weights < 0).any()
ornot np.isclose(weights.sum(), 1.0)
    ):
raise ValueError(
"目标权重必须非负且合计为1"
        )

    nav, _ = calculate_account_snapshot(
        account,
        prices,
    )

    target_weights = dict(
        zip(
            target["symbol"].astype(str),
            weights,
        )
    )

    all_symbols = sorted(
        set(target_weights)
        | set(account.positions)
    )

    rows = []

for symbol in all_symbols:

if symbol notin prices:
raise ValueError(
f"缺少{symbol}的价格"
            )

        price = float(prices[symbol])

ifnot np.isfinite(price) or price <= 0:
raise ValueError(
f"{symbol}价格不合法"
            )

        current_qty = (
            account.positions.get(
                symbol, 0
            )
        )

        weight = target_weights.get(
            symbol, 0.0
        )

        target_amount = nav * weight

# 目标持仓按整手向下取整
        target_qty = (
            math.floor(
                target_amount
                / (price * lot_size)
            ) * lot_size
        )

        delta = target_qty - current_qty

if delta > 0:
            side = "BUY"
elif delta < 0:
            side = "SELL"
else:
            side = "HOLD"

        rows.append({
"symbol": symbol,
"price": price,
"current_quantity": current_qty,
"target_weight": weight,
"target_quantity": target_qty,
"order_quantity": abs(delta),
"side": side,
        })

return pd.DataFrame(rows)

5.1 为什么要使用 dataclass？

我们之前学习过 C++，可以把它理解成一个简单的账户状态结构体。

structAccount {
double cash;
std::map<std::string, int> positions;
};

Python 中：

@dataclass(frozen=True)
classAccount:
    cash: float
    positions: dict[str, int]

frozen=True可以阻止重新给 cash或 positions字段赋值，但不能阻止修改字典内部内容。

例如：

account.positions["A"] = 100

仍可能修改字典。

因此，今天把 Account作为不主动修改的输入快照使用。后续实现正式账户引擎时，可以采用不可变映射或明确的状态更新接口。

5.2 这段代码为什么还不是交易执行引擎？

假设：

当前现金：0元
持有A股票：100000元

目标：
A 0%
B 100%

程序会生成：

SELL A
BUY B

但在真实交易中，必须先确认 A 的卖出成交，再根据实际到账可用资金安排 B 的买入。

当前函数只负责生成目标交易意图。

它没有模拟成交顺序和资金约束，因此不能直接把生成的买入指令全部执行。

这是我们 Day34 回测引擎需要解决的问题。

六、【工程入口代码】scripts/build_equal_weight.py

今天我们先从 Day32 的 LightGBM Top-K 选股信号中选择一个日期，构建示例账户。

这里使用教学模拟价格，不把它们当成真实历史行情。

创建：

scripts/build_equal_weight.py

# ============================================================
# Day33：等权目标组合构建入口
# 正式输出：lightgbm_target.parquet
# 演示输出：orders_demo / account_demo（不可用于历史回测）
# ============================================================

from pathlib import Path

import pandas as pd

from src.portfolio.equal_weight import (
    Account,
    build_equal_weight_target,
    calculate_account_snapshot,
    generate_order_intents,
)


SIGNAL_PATH = Path(
"data/signals/lightgbm_topk_test.parquet"
)

OUTPUT_DIR = Path(
"data/portfolios"
)


defmain():

    OUTPUT_DIR.mkdir(
        parents=True,
        exist_ok=True,
    )

ifnot SIGNAL_PATH.exists():
raise FileNotFoundError(
            SIGNAL_PATH
        )

    signals = pd.read_parquet(
        SIGNAL_PATH
    )

    target = build_equal_weight_target(
        signals
    )

    target.to_parquet(
        OUTPUT_DIR
        / "lightgbm_target.parquet",
        index=False,
    )

    latest_date = target[
"signal_date"
    ].max()

    latest_target = target.loc[
        target["signal_date"]
        == latest_date
    ].copy()

    print("\n最新一期目标组合：")

    print(
        latest_target.to_string(
            index=False
        )
    )

# 仅供组合构建功能验证。
# 不代表真实A股历史价格。
    demo_prices = {
        symbol: float(
10 + i * 5
        )
for i, symbol in enumerate(
            latest_target["symbol"]
        )
    }

    account = Account(
        cash=100000.0,
        positions={},
    )

    nav, snapshot = (
        calculate_account_snapshot(
            account,
            demo_prices,
        )
    )

    print("\n初始账户净值：")
    print(nav)

    orders = generate_order_intents(
        latest_target,
        account,
        demo_prices,
        lot_size=100,
    )

    orders.to_csv(
        OUTPUT_DIR
        / "lightgbm_orders_demo.csv",
        index=False,
        encoding="utf-8-sig",
    )

    snapshot.to_csv(
        OUTPUT_DIR
        / "lightgbm_account_demo.csv",
        index=False,
        encoding="utf-8-sig",
    )

    print("\n计划订单：")
    print(
        orders.to_string(index=False)
    )

    planned_buy_amount = (
        orders.loc[
            orders["side"] == "BUY",
"order_quantity",
        ]
        * orders.loc[
            orders["side"] == "BUY",
"price",
        ]
    ).sum()

    print(
"\n计划买入金额：",
        planned_buy_amount,
    )

    print(
"未考虑费用的剩余现金：",
        account.cash - planned_buy_amount,
    )

    print(
"\n注意：订单尚未成交，"
"账户实际持仓仍为空。"
    )


if __name__ == "__main__":
    main()

运行：

python -m scripts.build_equal_weight

成功后，检查：

ls data/portfolios

今天生成的 lightgbm_target.parquet是正式的研究目标组合数据；orders_demo.csv和 account_demo.csv只是功能演示，不能用于历史回测收益计算。

七、从目标组合到实际持仓：还缺少什么？

这是 Day33 必须明确的边界。

假设目标权重已经确定，但我们还没有完成：

• 根据历史真实价格确定可买数量。
• 检查股票是否停牌。
• 检查涨跌停限制。
• 检查买卖手续费。
• 检查印花税。
• 检查卖出股票是否满足 T+1 规则。
• 检查订单是否成交。
• 更新现金和持仓。

因此，我们的系统目前是：

模型预测
    ↓
Top-K选股
    ↓
目标权重
    ↓
计划订单
    ↓
【尚未实现】
真实成交模拟
    ↓
账户状态更新
    ↓
净值曲线

其中，今天已经完成前三个环节和基础计划订单。

Day34 才会正式研究回测。

八、为什么要区分目标权重和实际权重？

举个例子。

假设初始组合：

经过一段时间，A 上涨 20%，其他股票价格不变。

初始总资产 100 万元，A 初始市值 20 万元。

上涨后：

组合总资产：

此时 A 的实际权重：

其余股票：

这意味着：

即使选股集合完全不变，价格变化也会导致组合权重偏离目标。

因此，后续再平衡策略需要决定：

• 每天恢复等权？
• 每周恢复等权？
• 每月恢复等权？
• 还是只有权重偏离超过阈值才交易？

再平衡越频繁，通常意味着更高的交易成本。

但也可能更及时地控制风险。

这需要回测验证，不能仅凭直觉选择。

九、【自动化测试】tests/test_equal_weight.py

创建：

tests/test_equal_weight.py

完整代码：

# ============================================================
# Day33：等权组合构建单元测试
# ============================================================

import pandas as pd
import pytest

from src.portfolio.equal_weight import (
    Account,
    build_equal_weight_target,
    calculate_account_snapshot,
    generate_order_intents,
)


deftest_equal_weight_target():

    signals = pd.DataFrame({
"signal_date": [
"2025-01-02",
"2025-01-02",
"2025-01-02",
"2025-01-03",
"2025-01-03",
        ],
"symbol": [
"A", "B", "C",
"A", "D",
        ],
    })

    result = build_equal_weight_target(
        signals
    )

    first = result.loc[
        result["signal_date"]
        == pd.Timestamp("2025-01-02")
    ]

    second = result.loc[
        result["signal_date"]
        == pd.Timestamp("2025-01-03")
    ]

assert len(first) == 3
assert len(second) == 2

assert first[
"target_weight"
    ].iloc[0] == pytest.approx(1 / 3)

assert second[
"target_weight"
    ].iloc[0] == pytest.approx(0.5)


deftest_account_snapshot():

    account = Account(
        cash=10000.0,
        positions={
"A": 100,
"B": 200,
        },
    )

    prices = {
"A": 20.0,
"B": 30.0,
    }

    nav, snapshot = (
        calculate_account_snapshot(
            account,
            prices,
        )
    )

assert nav == pytest.approx(
18000.0
    )

assert snapshot[
"actual_weight"
    ].sum() == pytest.approx(
8000 / 18000
    )


deftest_generate_buy_orders():

    target = pd.DataFrame({
"symbol": ["A", "B"],
"target_weight": [0.5, 0.5],
    })

    account = Account(
        cash=100000.0,
        positions={},
    )

    prices = {
"A": 20.0,
"B": 25.0,
    }

    orders = generate_order_intents(
        target,
        account,
        prices,
    )

    a = orders.loc[
        orders["symbol"] == "A"
    ].iloc[0]

    b = orders.loc[
        orders["symbol"] == "B"
    ].iloc[0]

assert a["side"] == "BUY"
assert b["side"] == "BUY"

assert a["order_quantity"] == 2500
assert b["order_quantity"] == 2000


deftest_generate_sell_order():

    target = pd.DataFrame({
"symbol": ["B"],
"target_weight": [1.0],
    })

    account = Account(
        cash=0.0,
        positions={
"A": 100,
        },
    )

    prices = {
"A": 20.0,
"B": 10.0,
    }

    orders = generate_order_intents(
        target,
        account,
        prices,
    )

    a = orders.loc[
        orders["symbol"] == "A"
    ].iloc[0]

assert a["side"] == "SELL"
assert a["order_quantity"] == 100


deftest_invalid_weights():

    target = pd.DataFrame({
"symbol": ["A", "B"],
"target_weight": [0.4, 0.4],
    })

    account = Account(
        cash=100000.0,
        positions={},
    )

with pytest.raises(ValueError):

        generate_order_intents(
            target,
            account,
            {
"A": 10.0,
"B": 20.0,
            },
        )

运行：

pytest -q tests/test_equal_weight.py

再运行：

python -m scripts.build_equal_weight

十、今天的高质量练习题

理论题

题1：目标权重与实际权重有什么区别？为什么二者可能不相等？

题2：一个 5 只股票的等权组合，初始资金 100 万元，每只股票分配多少？

题3：如果一只股票价格为 36.80 元，目标分配资金为 20000 元，按照 100 股整手买入，最多可以买多少股？剩余多少现金？

题4：为什么组合净值上涨之后，原来的等权组合可能变成非等权？

题5：为什么订单意图不能直接当作实际成交？

编程题

题6：修改 build_equal_weight_target()，支持 10% 的现金预留。

例如：

A 18%
B 18%
C 18%
D 18%
E 18%
现金 10%

题7：编写一个 calculate_weight_drift()函数，比较目标权重与实际权重。

题8：设计一个再平衡阈值，例如：

rebalance_threshold = 0.03

只有目标权重与实际权重的偏差超过 3 个百分点，才产生调整意图。

题9：在不考虑手续费的前提下，计算整手买入造成的剩余现金，并计算实际股票权重。

题10：思考：如果账户已经持有某只股票，但当天因为 T+1 限制不能卖出，应该如何处理目标组合？

十一、部分练习参考答案

题3：整手买入

买入金额：

剩余现金：

题6：保留 10% 现金

可以把 build_equal_weight_target()扩展为：

defbuild_equal_weight_target(
    signals: pd.DataFrame,
    cash_reserve: float = 0.0,
) -> pd.DataFrame:

ifnot0 <= cash_reserve < 1:
raise ValueError(
"cash_reserve必须在[0,1)之间"
        )

# 省略原有输入验证代码

    df = signals.copy()

    counts = df.groupby(
"signal_date"
    )["symbol"].transform("size")

    df["target_weight"] = (
        (1.0 - cash_reserve) / counts
    )

return df

这里有一个配套修改：今天 generate_order_intents()要求目标股票权重合计为 1。如果支持现金预留，也必须将该验证条件修改为允许权重合计小于等于 1，并明确剩余部分代表目标现金权重。

这两个函数需要一起调整，不能只改其中一个。

题7：计算权重偏离

defcalculate_weight_drift(
    target: pd.DataFrame,
    snapshot: pd.DataFrame,
) -> pd.DataFrame:

    result = target.merge(
        snapshot[
            ["symbol", "actual_weight"]
        ],
        on="symbol",
        how="outer",
    )

    result["target_weight"] = (
        result["target_weight"].fillna(0)
    )

    result["actual_weight"] = (
        result["actual_weight"].fillna(0)
    )

    result["weight_drift"] = (
        result["target_weight"]
        - result["actual_weight"]
    )

return result

注意，目标组合中不存在、但账户仍持有的股票，也必须纳入比较。

十二、Day33 工程验收

十三、Day33 总结

今天，我们完成了从 Top-K 选股到等权组合构建的工程衔接。

目前 quant_ml已经具备：

真实A股数据
     │
     ▼
特征工程与收益标签
     │
     ▼
Linear / LightGBM / CatBoost
     │
     ▼
IC / RankIC / ICIR
     │
     ▼
Feature Importance / SHAP
     │
     ▼
Top-K模型选股
     │
     ▼
等权目标组合
     │
     ▼
当前账户净值与权重
     │
     ▼
计划订单
     │
     ▼
Day34：历史回测引擎

今天最重要的一句话：

模型决定想买什么，组合构建决定资金如何分配，而交易执行决定最终真正持有什么。三者必须严格分离。

下一课：Day34——策略回测

按照原定计划，Day34 将进入完整的策略回测：

• 历史行情与选股信号按时间对齐。
• T 日盘后生成信号，T+1 执行。
• 现金、持仓和组合净值更新。
• 手续费、印花税、滑点。
• 调仓与交易日志。
• 策略收益、最大回撤、夏普比率。
• 对照基准与样本外评价。

Day35 则会把这一阶段的模块整合为完整的 ML 选股系统。