264 lines
9.8 KiB
Python
264 lines
9.8 KiB
Python
"""持仓规模优化模块。
|
||
|
||
支持多种仓位管理方法：
|
||
- equal_weight: 等权分配（默认）
|
||
- kelly: Kelly公式（需要历史胜率和盈亏比）
|
||
- volatility_target: 波动率目标（根据股票波动率调整仓位）
|
||
|
||
使用方法：
|
||
position_sizer = PositionSizing(method="equal_weight", max_positions=2)
|
||
|
||
# 在策略中使用
|
||
if position_sizer.can_open(ts_code, cash, price):
|
||
shares = position_sizer.calc_shares(ts_code, cash, price, df)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Dict, Optional
|
||
|
||
import pandas as pd
|
||
|
||
from utils.logger import setup_logger
|
||
|
||
logger = setup_logger(__name__)
|
||
|
||
|
||
class PositionSizing:
|
||
"""持仓规模优化管理器。
|
||
|
||
支持多种方法：
|
||
- equal_weight: 等权分配
|
||
- kelly: Kelly公式
|
||
- volatility_target: 波动率目标
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
method: str = "equal_weight",
|
||
max_positions: int = 2,
|
||
kelly_risk_free: float = 0.03,
|
||
kelly_max_fraction: float = 0.25,
|
||
volatility_target: float = 0.15,
|
||
volatility_window: int = 20,
|
||
):
|
||
"""初始化持仓规模管理器。
|
||
|
||
参数：
|
||
method: 仓位管理方法 ("equal_weight", "kelly", "volatility_target")
|
||
max_positions: 最大持仓数
|
||
kelly_risk_free: Kelly公式的无风险利率
|
||
kelly_max_fraction: Kelly公式的最大仓位比例（防止过度杠杆）
|
||
volatility_target: 目标波动率（年化）
|
||
volatility_window: 计算波动率的窗口期
|
||
"""
|
||
self.method = method
|
||
self.max_positions = max_positions
|
||
self.kelly_risk_free = kelly_risk_free
|
||
self.kelly_max_fraction = kelly_max_fraction
|
||
self.volatility_target = volatility_target
|
||
self.volatility_window = volatility_window
|
||
|
||
# 记录每只股票的历史交易统计（用于Kelly公式）
|
||
self.trade_stats: Dict[str, Dict] = {}
|
||
|
||
def can_open(self, ts_code: str, cash: float, price: float, current_positions: int) -> bool:
|
||
"""判断是否可以开新仓位。
|
||
|
||
参数：
|
||
ts_code: 股票代码
|
||
cash: 当前可用现金
|
||
price: 当前价格
|
||
current_positions: 当前持仓数量
|
||
|
||
返回：
|
||
bool: 是否可以开仓
|
||
"""
|
||
# 检查仓位是否已满
|
||
if current_positions >= self.max_positions:
|
||
return False
|
||
|
||
# 检查资金是否足够买入至少 100 股
|
||
min_cost = price * 100
|
||
if cash < min_cost:
|
||
return False
|
||
|
||
return True
|
||
|
||
def calc_shares(
|
||
self,
|
||
ts_code: str,
|
||
cash: float,
|
||
price: float,
|
||
remain_slots: int,
|
||
df: Optional[pd.DataFrame] = None,
|
||
) -> int:
|
||
"""计算应该买入的股数。
|
||
|
||
参数：
|
||
ts_code: 股票代码
|
||
cash: 当前可用现金
|
||
price: 当前价格
|
||
remain_slots: 剩余可用仓位数
|
||
df: 股票历史数据（用于计算波动率等指标）
|
||
|
||
返回：
|
||
int: 买入股数（已取整到100的整数倍）
|
||
"""
|
||
if self.method == "equal_weight":
|
||
return self._equal_weight_shares(cash, price, remain_slots)
|
||
elif self.method == "kelly":
|
||
return self._kelly_shares(ts_code, cash, price, remain_slots)
|
||
elif self.method == "volatility_target":
|
||
return self._volatility_target_shares(ts_code, cash, price, remain_slots, df)
|
||
else:
|
||
logger.warning(f"未知的仓位管理方法: {self.method}，使用等权分配")
|
||
return self._equal_weight_shares(cash, price, remain_slots)
|
||
|
||
def _equal_weight_shares(self, cash: float, price: float, remain_slots: int) -> int:
|
||
"""等权分配：平均分配现金到剩余仓位。
|
||
|
||
例如：现金100万，剩余2个仓位，每个仓位分配50万
|
||
"""
|
||
if remain_slots <= 0:
|
||
return 0
|
||
|
||
cash_per_stock = cash / remain_slots
|
||
shares = int(cash_per_stock // price)
|
||
|
||
# A股规则：向下取整到100的整数倍
|
||
shares = (shares // 100) * 100
|
||
|
||
return shares
|
||
|
||
def _kelly_shares(self, ts_code: str, cash: float, price: float, remain_slots: int) -> int:
|
||
"""Kelly公式：根据历史胜率和盈亏比计算最优仓位。
|
||
|
||
Kelly% = (胜率 * 盈亏比 - 败率) / 盈亏比
|
||
|
||
注意：需要积累一定的交易历史才能准确计算。
|
||
如果没有历史数据，回退到等权分配。
|
||
"""
|
||
stats = self.trade_stats.get(ts_code)
|
||
|
||
if stats is None or stats.get("total_trades", 0) < 10:
|
||
# 交易次数不足，使用等权分配
|
||
logger.debug(f"{ts_code} 历史交易不足，使用等权分配")
|
||
return self._equal_weight_shares(cash, price, remain_slots)
|
||
|
||
win_rate = stats.get("win_rate", 0.5)
|
||
avg_win = stats.get("avg_win", 0.05)
|
||
avg_loss = stats.get("avg_loss", 0.05)
|
||
|
||
if avg_loss <= 0:
|
||
avg_loss = 0.01 # 避免除零
|
||
|
||
profit_loss_ratio = avg_win / avg_loss
|
||
kelly_fraction = (win_rate * profit_loss_ratio - (1 - win_rate)) / profit_loss_ratio
|
||
|
||
# Kelly公式可能给出负值或过大值，需要限制
|
||
kelly_fraction = max(0, min(kelly_fraction, self.kelly_max_fraction))
|
||
|
||
# 考虑剩余仓位数，分配资金
|
||
cash_per_stock = cash / remain_slots
|
||
kelly_cash = cash_per_stock * kelly_fraction
|
||
|
||
shares = int(kelly_cash // price)
|
||
shares = (shares // 100) * 100
|
||
|
||
logger.debug(
|
||
f"{ts_code} Kelly仓位: {kelly_fraction*100:.2f}%, "
|
||
f"胜率={win_rate*100:.1f}%, 盈亏比={profit_loss_ratio:.2f}"
|
||
)
|
||
|
||
return shares
|
||
|
||
def _volatility_target_shares(
|
||
self,
|
||
ts_code: str,
|
||
cash: float,
|
||
price: float,
|
||
remain_slots: int,
|
||
df: Optional[pd.DataFrame],
|
||
) -> int:
|
||
"""波动率目标：根据股票波动率调整仓位。
|
||
|
||
波动率高的股票减小仓位，波动率低的股票增大仓位。
|
||
目标：使每个仓位的波动率贡献接近目标波动率。
|
||
"""
|
||
if df is None or df.empty:
|
||
logger.debug(f"{ts_code} 缺少历史数据，使用等权分配")
|
||
return self._equal_weight_shares(cash, price, remain_slots)
|
||
|
||
# 计算历史波动率（日收益率标准差 * sqrt(252)）
|
||
if "close" not in df.columns or len(df) < self.volatility_window:
|
||
logger.debug(f"{ts_code} 数据不足，使用等权分配")
|
||
return self._equal_weight_shares(cash, price, remain_slots)
|
||
|
||
returns = df["close"].pct_change().dropna()
|
||
|
||
if len(returns) < self.volatility_window:
|
||
logger.debug(f"{ts_code} 数据不足，使用等权分配")
|
||
return self._equal_weight_shares(cash, price, remain_slots)
|
||
|
||
# 使用最近 volatility_window 天的数据计算波动率
|
||
recent_volatility = returns.tail(self.volatility_window).std() * (252 ** 0.5)
|
||
|
||
if recent_volatility <= 0:
|
||
logger.debug(f"{ts_code} 波动率为0，使用等权分配")
|
||
return self._equal_weight_shares(cash, price, remain_slots)
|
||
|
||
# 仓位调整因子 = 目标波动率 / 实际波动率
|
||
volatility_factor = self.volatility_target / recent_volatility
|
||
volatility_factor = max(0.5, min(volatility_factor, 2.0)) # 限制在 [0.5, 2.0]
|
||
|
||
# 基础等权分配 * 波动率因子
|
||
base_shares = self._equal_weight_shares(cash, price, remain_slots)
|
||
adjusted_shares = int(base_shares * volatility_factor)
|
||
adjusted_shares = (adjusted_shares // 100) * 100
|
||
|
||
logger.debug(
|
||
f"{ts_code} 波动率调整: 实际={recent_volatility*100:.2f}%, "
|
||
f"目标={self.volatility_target*100:.2f}%, 因子={volatility_factor:.2f}"
|
||
)
|
||
|
||
return adjusted_shares
|
||
|
||
def update_trade_stats(self, ts_code: str, profit_pct: float) -> None:
|
||
"""更新交易统计（用于Kelly公式）。
|
||
|
||
参数：
|
||
ts_code: 股票代码
|
||
profit_pct: 本次交易盈亏比例（正数为盈利，负数为亏损）
|
||
"""
|
||
if ts_code not in self.trade_stats:
|
||
self.trade_stats[ts_code] = {
|
||
"total_trades": 0,
|
||
"wins": 0,
|
||
"losses": 0,
|
||
"total_win": 0.0,
|
||
"total_loss": 0.0,
|
||
}
|
||
|
||
stats = self.trade_stats[ts_code]
|
||
stats["total_trades"] += 1
|
||
|
||
if profit_pct > 0:
|
||
stats["wins"] += 1
|
||
stats["total_win"] += profit_pct
|
||
else:
|
||
stats["losses"] += 1
|
||
stats["total_loss"] += abs(profit_pct)
|
||
|
||
# 计算平均值
|
||
stats["win_rate"] = stats["wins"] / stats["total_trades"]
|
||
stats["avg_win"] = stats["total_win"] / stats["wins"] if stats["wins"] > 0 else 0.05
|
||
stats["avg_loss"] = stats["total_loss"] / stats["losses"] if stats["losses"] > 0 else 0.05
|
||
|
||
def get_stats(self, ts_code: str) -> Optional[Dict]:
|
||
"""获取某只股票的交易统计。"""
|
||
return self.trade_stats.get(ts_code)
|
||
|
||
def clear_stats(self) -> None:
|
||
"""清空所有交易统计。"""
|
||
self.trade_stats.clear()
|