引言：为什么需要打分制量化模型
在股票投资中，情绪化交易是导致投资者亏损的主要原因之一。恐惧、贪婪、从众心理等情绪因素常常让投资者在市场高点追涨，在市场低点割肉。根据行为金融学研究，超过80%的个人投资者因为情绪化决策而跑输大盘。打分制量化模型通过建立客观、可重复的评估体系，将投资决策从情绪驱动转变为数据驱动，从而实现长期稳定收益。
打分制量化模型的核心优势在于：
- 客观性：消除主观判断偏差，所有决策基于预设规则
- 可回溯性：所有评分历史可追溯，便于策略优化
- 纪律性：强制执行买卖纪律，避免冲动交易
- 可扩展性：可同时监控数百只股票，提高决策效率
一、打分制量化模型的核心框架
1.1 模型设计原则
构建打分制量化模型需要遵循以下核心原则：
原则一：多维度综合评估 单一指标无法全面反映股票价值，需要从基本面、技术面、市场情绪等多个维度综合评估。例如，一只股票可能基本面优秀但技术面走坏，或者技术面强势但估值过高。
原则二：权重动态调整 不同市场环境下各维度的重要性不同。牛市中技术面权重可适当提高，熊市中则应更重视基本面和估值指标。
原则三：阈值管理 设置明确的买入、卖出、持有阈值，避免模糊决策。例如，总分≥80分买入，≤40分卖出，40-80分持有。
原则四：风险控制优先 任何投资策略都应将风险控制放在首位，设置最大回撤限制、单只股票仓位上限等。
1.2 评分维度设计
一个完整的打分制量化模型通常包含以下维度：
基本面维度（权重30%）
- 盈利能力：ROE、ROA、毛利率、净利率
- 成长性：营收增长率、净利润增长率
- 财务健康度：资产负债率、流动比率、经营性现金流
- 估值水平：PE、PB、PS、PEG
技术面维度（权重30%）
- 趋势指标：均线系统（MA5/MA20/MA60）、MACD
- 动量指标：RSI、ROC、布林带位置
- 成交量：量比、换手率、成交量趋势
- 波动率：ATR、历史波动率
市场情绪维度（权重20%）
- 资金流向：主力资金净流入、北向资金持仓变化
- 市场关注度：研报数量、机构评级变化
- 舆情分析：新闻情绪指数、社交媒体热度
风险维度（权重20%）
- 波动风险：最大回撤、夏普比率
- 流动性风险：日均成交量、冲击成本
- 集中度风险：行业集中度、个股相关性
二、实战模型构建详解
2.1 数据准备与清洗
在构建模型前，需要准备高质量的数据源。以下是获取股票数据的Python示例：
import pandas as pd
import numpy as np
import akshare as ak
import warnings
warnings.filterwarnings('ignore')
class StockDataFetcher:
"""股票数据获取器"""
def __init__(self):
self.fetcher = ak
def get_stock_daily(self, symbol, start_date, end_date):
"""获取日线数据"""
try:
df = ak.stock_zh_a_hist(symbol=symbol, period="daily",
start_date=start_date, end_date=end_date)
df.columns = ['date', 'open', 'close', 'high', 'low', 'volume',
'turnover', 'amplitude', 'change_pct', 'change_amount', 'turnover_rate']
return df
except Exception as e:
print(f"获取{symbol}数据失败: {e}")
return None
def get_finance_data(self, symbol):
"""获取财务数据"""
try:
# 获取最新财务指标
finance_df = ak.stock_financial_analysis_indicator(symbol)
return finance_df
except Exception as e:
print(f"获取{symbol}财务数据失败: {e}")
return None
def get_market_sentiment(self):
"""获取市场情绪数据"""
try:
# 获取北向资金数据
north_flow = ak.stock_hk_ggt_sina()
# 获取融资融券数据
margin_data = ak.stock_margin_sse()
return {'north_flow': north_flow, 'margin_data': margin_data}
except Exception as e:
print(f"获取市场情绪数据失败: {e}")
return None
# 使用示例
fetcher = StockDataFetcher()
# 获取贵州茅台数据
df_mt = fetcher.get_stock_daily('600519', '20230101', '20240101')
print(df_mt.head())
2.2 评分指标计算模块
接下来构建核心的评分计算模块，每个指标都有明确的评分标准：
class ScoringMetrics:
"""评分指标计算类"""
def __init__(self):
pass
@staticmethod
def calculate_roe_score(roe):
"""ROE评分（0-25分）"""
if roe >= 20:
return 25
elif roe >= 15:
return 20
elif roe >= 10:
return 15
elif roe >= 5:
return 10
else:
return 5
@staticmethod
def calculate_pe_score(pe, industry_avg_pe):
"""PE估值评分（0-25分）"""
if pe <= industry_avg_pe * 0.7:
return 25 # 显著低估
elif pe <= industry_avg_pe * 0.9:
return 20 # 略低估
elif pe <= industry_avg_pe * 1.1:
return 15 # 合理
elif pe <= industry_avg_pe * 1.5:
return 10 # 略高估
else:
return 5 # 严重高估
@staticmethod
def calculate_growth_score(revenue_growth, profit_growth):
"""成长性评分（0-25分）"""
growth_score = 0
# 营收增长评分（12.5分）
if revenue_growth >= 30:
growth_score += 12.5
elif revenue_growth >= 20:
growth_score += 10
elif revenue_growth >= 10:
growth_score += 7.5
elif revenue_growth >= 5:
growth_score += 5
else:
growth_score += 2.5
# 净利润增长评分（12.5分）
if profit_growth >= 30:
growth_score += 12.5
elif profit_growth >= 20:
growth_score += 10
elif profit_growth >= 10:
growth_score += 7.5
elif profit_growth >= 5:
growth_score += 5
else:
growth_score += 2.5
return growth_score
@staticmethod
def calculate_technical_score(df, window=20):
"""技术面评分（0-30分）"""
if len(df) < window:
return 0
score = 0
close = df['close'].values
# 均线系统（10分）
ma5 = np.mean(close[-5:])
ma20 = np.mean(close[-20:])
ma60 = np.mean(close[-60:]) if len(close) >= 60 else ma20
if close[-1] > ma5 > ma20 > ma60:
score += 10 # 多头排列
elif close[-1] > ma20 and ma20 > ma60:
score += 7
elif close[-1] > ma60:
score += 5
else:
score += 2
# RSI动量（10分）
delta = np.diff(close)
gain = np.where(delta > 0, delta, 0)
loss = np.where(delta < 0, -delta, 0)
avg_gain = np.mean(gain[-14:])
avg_loss = np.mean(loss[-14:])
if avg_loss == 0:
rs = 100
else:
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
if 30 <= rsi <= 70:
score += 10 # 健康区间
elif rsi < 30:
score += 5 # 超卖
else:
score += 5 # 超买
# 成交量（10分）
volume = df['volume'].values
if len(volume) >= 20:
avg_volume = np.mean(volume[-20:])
current_volume = volume[-1]
if current_volume > avg_volume * 1.5:
score += 10 # 放量上涨
elif current_volume > avg_volume:
score += 7
else:
score += 5
return score
@staticmethod
def calculate_risk_score(df, window=60):
"""风险评分（0-20分）"""
if len(df) < window:
return 0
# 计算最大回撤（10分）
close = df['close'].values
running_max = np.maximum.accumulate(close)
drawdown = (close - running_max) / running_max
max_drawdown = np.min(drawdown)
if max_drawdown > -0.2:
risk_score = 10
elif max_drawdown > -0.3:
risk_score = 7
elif max_drawdown > -0.4:
risk_score = 5
else:
risk_score = 2
# 计算波动率（10分）
returns = np.diff(np.log(close))
volatility = np.std(returns) * np.sqrt(252) # 年化波动率
if volatility < 0.2:
risk_score += 10
elif volatility < 0.3:
risk_score += 7
elif volatility < 0.4:
risk_score += 5
else:
risk_score += 2
return risk_score
# 使用示例
metrics = ScoringMetrics()
# 假设我们有贵州茅台的财务数据
roe = 30.5 # ROE 30.5%
pe = 35 # PE 35倍
industry_pe = 40 # 行业平均PE 40倍
revenue_growth = 16.5 # 营收增长16.5%
profit_growth = 19.2 # 净利润增长19.2%
# 计算基本面得分
fundamental_score = (
metrics.calculate_roe_score(roe) +
metrics.calculate_pe_score(pe, industry_pe) +
metrics.calculate_growth_score(revenue_growth, profit_growth)
)
print(f"基本面得分: {fundamental_score}")
2.3 综合评分模型
将各维度得分整合，构建完整的评分模型：
class StockScoringModel:
"""股票综合评分模型"""
def __init__(self):
self.metrics = ScoringMetrics()
# 设置各维度权重
self.weights = {
'fundamental': 0.30, # 基本面
'technical': 0.30, # 技术面
'sentiment': 0.20, # 市场情绪
'risk': 0.20 # 风险
}
# 评分阈值
self.buy_threshold = 80
self.sell_threshold = 40
self.hold_threshold_low = 40
self.hold_threshold_high = 80
def calculate_fundamental_score(self, finance_data):
"""计算基本面得分（满分30分）"""
if finance_data is None or finance_data.empty:
return 0
# 获取最新数据
latest = finance_data.iloc[-1]
# ROE评分（10分）
roe = latest.get('roe', 0)
roe_score = self.metrics.calculate_roe_score(roe) * 0.4
# PE估值评分（10分）
pe = latest.get('pe', 999)
industry_pe = latest.get('industry_pe', 40)
pe_score = self.metrics.calculate_pe_score(pe, industry_pe) * 0.4
# 成长性评分（10分）
revenue_growth = latest.get('revenue_growth', 0)
profit_growth = latest.get('profit_growth', 0)
growth_score = self.metrics.calculate_growth_score(revenue_growth, profit_growth) * 0.4
return roe_score + pe_score + growth_score
def calculate_technical_score(self, price_data):
"""计算技术面得分（满分30分）"""
if price_data is None or price_data.empty:
return 0
return self.metrics.calculate_technical_score(price_data)
def calculate_sentiment_score(self, symbol, sentiment_data):
"""计算市场情绪得分（满分20分）"""
score = 0
# 北向资金流向（10分）
if sentiment_data and 'north_flow' in sentiment_data:
north_flow = sentiment_data['north_flow']
if symbol in north_flow.index:
flow_change = north_flow.loc[symbol, 'change']
if flow_change > 0:
score += 10
elif flow_change > -0.5:
score += 7
else:
score += 3
# 融资融券数据（10分）
if sentiment_data and 'margin_data' in sentiment_data:
margin_data = sentiment_data['margin_data']
if symbol in margin_data.index:
margin_balance = margin_data.loc[symbol, 'margin_balance_change']
if margin_balance > 0:
score += 10
elif margin_balance > -0.5:
score += 7
else:
score += 3
return score
def calculate_risk_score(self, price_data):
"""计算风险得分（满分20分）"""
if price_data is None or price_data.empty:
return 0
return self.metrics.calculate_risk_score(price_data)
def calculate_total_score(self, symbol, price_data, finance_data, sentiment_data):
"""计算综合得分"""
# 计算各维度得分
fundamental_score = self.calculate_fundamental_score(finance_data)
technical_score = self.calculate_technical_score(price_data)
sentiment_score = self.calculate_sentiment_score(symbol, sentiment_data)
risk_score = self.calculate_risk_score(price_data)
# 加权汇总
total_score = (
fundamental_score * self.weights['fundamental'] +
technical_score * self.weights['technical'] +
sentiment_score * self.weights['sentiment'] +
risk_score * self.weights['risk']
)
# 生成决策信号
if total_score >= self.buy_threshold:
signal = 'BUY'
elif total_score <= self.sell_threshold:
signal = 'SELL'
else:
signal = 'HOLD'
return {
'symbol': symbol,
'total_score': round(total_score, 2),
'fundamental_score': round(fundamental_score, 2),
'technical_score': round(technical_score, 2),
'sentiment_score': round(sentiment_score, 2),
'risk_score': round(risk_score, 1),
'signal': signal,
'timestamp': pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')
}
# 使用示例
model = StockScoringModel()
# 模拟数据
price_data = fetcher.get_stock_daily('600519', '20230101', '2240101')
finance_data = fetcher.get_finance_data('600519')
sentiment_data = fetcher.get_market_sentiment()
# 计算评分
result = model.calculate_total_score('600519', price_data, finance_data, sentiment_data)
print(f"综合评分结果: {result}")
2.4 回测系统实现
完整的量化模型必须经过严格的回测验证。以下是回测系统代码：
class BacktestSystem:
"""回测系统"""
def __init__(self, initial_capital=100000):
self.initial_capital = initial_capital
self.capital = initial_capital
self.position = {} # 持仓：symbol -> shares
self.trade_log = [] # 交易记录
self.daily_value = [] # 每日净值
def run_backtest(self, symbol_list, start_date, end_date, model):
"""运行回测"""
print(f"开始回测: {start_date} 至 {end_date}")
# 获取所有股票数据
stock_data = {}
for symbol in symbol_list:
price_data = fetcher.get_stock_daily(symbol, start_date, end_date)
finance_data = fetcher.get_finance_data(symbol)
if price_data is not None:
stock_data[symbol] = {
'price': price_data,
'finance': finance_data
}
# 获取市场情绪数据（每月更新）
sentiment_data = fetcher.get_market_sentiment()
# 按交易日迭代
all_dates = sorted(set(
pd.to_datetime(stock_data[symbol]['price']['date'])
for symbol in stock_data
))
for date in all_dates:
date_str = date.strftime('%Y-%m-%d')
daily_value = self.capital
# 评估每只股票
for symbol in stock_data:
price_data = stock_data[symbol]['price']
finance_data = stock_data[symbol]['finance']
# 获取当日数据
daily_price = price_data[price_data['date'] == date_str]
if daily_price.empty:
continue
# 计算评分（每月第一个交易日）
if date.day == 1 or not self.trade_log:
score_result = model.calculate_total_score(
symbol, price_data, finance_data, sentiment_data
)
# 执行交易逻辑
current_price = daily_price['close'].values[0]
self._execute_trade(symbol, score_result, current_price)
# 计算持仓市值
if symbol in self.position:
shares = self.position[symbol]
daily_value += shares * current_price
self.daily_value.append({
'date': date_str,
'value': daily_value,
'return': (daily_value - self.initial_capital) / self.initial_capital
})
return self._calculate_performance_metrics()
def _execute_trade(self, symbol, score_result, price):
"""执行交易"""
signal = score_result['signal']
if signal == 'BUY' and symbol not in self.position:
# 计算可买入数量（每次使用10%资金）
buy_amount = self.capital * 0.1
shares = int(buy_amount / price)
if shares > 0:
cost = shares * price
self.capital -= cost
self.position[symbol] = shares
self.trade_log.append({
'date': score_result['timestamp'],
'symbol': symbol,
'action': 'BUY',
'price': price,
'shares': shares,
'cost': cost,
'score': score_result['total_score']
})
print(f"买入 {symbol}: {shares}股 @ {price}, 评分: {score_result['total_score']}")
elif signal == 'SELL' and symbol in self.position:
# 卖出全部持仓
shares = self.position[symbol]
revenue = shares * price
self.capital += revenue
del self.position[symbol]
self.trade_log.append({
'date': score_result['timestamp'],
'symbol': symbol,
'action': 'SELL',
'price': price,
'shares': shares,
'revenue': revenue,
'score': score_result['total_score']
})
print(f"卖出 {symbol}: {shares}股 @ {price}, 评分: {score_result['total_score']}")
def _calculate_performance_metrics(self):
"""计算回测绩效指标"""
if not self.daily_value:
return None
df = pd.DataFrame(self.daily_value)
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)
# 总收益率
total_return = df['return'].iloc[-1]
# 年化收益率
days = (df.index[-1] - df.index[0]).days
annual_return = (1 + total_return) ** (365 / days) - 1 if days > 0 else 0
# 最大回撤
running_max = df['value'].cummax()
drawdown = (df['value'] - running_max) / running_max
max_drawdown = drawdown.min()
# 夏普比率（假设无风险利率3%）
returns = df['value'].pct_change().dropna()
excess_returns = returns - 0.03 / 252
sharpe_ratio = excess_returns.mean() / excess_returns.std() * np.sqrt(252) if returns.std() > 0 else 0
# 胜率
trades = pd.DataFrame(self.trade_log)
if not trades.empty:
buy_trades = trades[trades['action'] == 'BUY']
sell_trades = trades[trades['action'] == 'SELL']
# 计算每笔交易盈亏
profits = []
for _, buy in buy_trades.iterrows():
sell = sell_trades[sell_trades['symbol'] == buy['symbol']].iloc[0]
profit = (sell['price'] - buy['price']) * buy['shares']
profits.append(profit)
win_rate = len([p for p in profits if p > 0]) / len(profits) if profits else 0
else:
win_rate = 0
return {
'初始资金': self.initial_capital,
'最终资金': self.capital,
'总收益率': f"{total_return:.2%}",
'年化收益率': f"{annual_return:.2%}",
'最大回撤': f"{max_drawdown:.2%}",
'夏普比率': f"{sharpe_ratio:.2f}",
'交易次数': len(self.trade_log),
'胜率': f"{win_rate:.2%}",
'当前持仓': self.position
}
# 使用示例
backtest = BacktestSystem(initial_capital=100000)
symbol_list = ['600519', '000858', '600036'] # 贵州茅台、五粮液、招商银行
results = backtest.run_backtest(symbol_list, '20230101', '20240101', model)
print("\n回测结果:")
for k, v in results.items():
print(f"{k}: {v}")
三、模型优化与风险管理
3.1 动态权重调整
根据市场环境动态调整权重，提高模型适应性：
class DynamicWeightOptimizer:
"""动态权重优化器"""
def __init__(self):
self.market_regime = 'normal' # bull, bear, normal
def detect_market_regime(self, market_index_data):
"""识别市场状态"""
if len(market_index_data) < 60:
return 'normal'
# 计算20日和60日均线
close = market_index_data['close'].values
ma20 = np.mean(close[-20:])
ma60 = np.mean(close[-60:])
# 计算60日收益率
returns_60 = (close[-1] - close[-60]) / close[-60]
if ma20 > ma60 and returns_60 > 0.1:
return 'bull'
elif ma20 < ma60 and returns_60 < -0.1:
return 'bear'
else:
return 'normal'
def get_dynamic_weights(self, market_regime):
"""根据市场状态返回权重"""
weights = {
'fundamental': 0.30,
'technical': 0.30,
'sentiment': 0.20,
'risk': 0.20
}
if market_regime == 'bull':
# 牛市：提高技术面和情绪权重
weights['technical'] = 0.35
weights['sentiment'] = 0.25
weights['fundamental'] = 0.25
weights['risk'] = 0.15
elif market_regime == 'bear':
# 熊市：提高基本面和风险权重
weights['fundamental'] = 0.40
weights['risk'] = 0.30
weights['technical'] = 0.15
weights['sentiment'] = 0.15
return weights
# 使用示例
optimizer = DynamicWeightOptimizer()
# 获取沪深300指数数据
market_index = fetcher.get_stock_daily('000300', '20230101', '20240101')
regime = optimizer.detect_market_regime(market_index)
dynamic_weights = optimizer.get_dynamic_weights(regime)
print(f"当前市场状态: {regime}")
print(f"动态权重: {dynamic_weights}")
3.2 风险控制模块
严格的风险控制是长期稳定收益的保障：
class RiskController:
"""风险控制器"""
def __init__(self, max_position_per_stock=0.15, max_drawdown_limit=0.2):
self.max_position_per_stock = max_position_per_stock # 单只股票最大仓位
self.max_drawdown_limit = max_drawdown_limit # 最大回撤限制
self.stop_loss_level = None # 动态止损线
def check_position_limit(self, symbol, position_value, total_capital):
"""检查仓位限制"""
position_ratio = position_value / total_capital
if position_ratio > self.max_position_per_stock:
return False, f"{symbol}仓位{position_ratio:.1%}超过限制{self.max_position_per_stock:.1%}"
return True, "仓位正常"
def check_drawdown_limit(self, current_value, peak_value):
"""检查回撤限制"""
drawdown = (peak_value - current_value) / peak_value
if drawdown > self.max_drawdown_limit:
return False, f"当前回撤{drawdown:.1%}超过限制{self.max_drawdown_limit:.1%}"
return True, "回撤正常"
def calculate_stop_loss(self, entry_price, current_price, atr, atr_multiplier=2):
"""动态止损计算"""
if self.stop_loss_level is None:
self.stop_loss_level = entry_price - atr * atr_multiplier
# 跟踪止损（只上调不下调）
new_stop = current_price - atr * atr_multiplier
if new_stop > self.stop_loss_level:
self.stop_loss_level = new_stop
return self.stop_loss_level
def check_liquidity(self, symbol, volume, avg_volume):
"""检查流动性"""
if volume < avg_volume * 0.5:
return False, f"{symbol}成交量过低，流动性不足"
return True, "流动性正常"
# 使用示例
risk_ctrl = RiskController(max_position_per_stock=0.15, max_drawdown_limit=0.2)
# 检查仓位
position_value = 20000
total_capital = 100000
ok, msg = risk_ctrl.check_position_limit('600519', position_value, total_capital)
print(f"仓位检查: {msg}")
# 动态止损
entry_price = 1800
current_price = 1850
atr = 30
stop_price = risk_ctrl.calculate_stop_loss(entry_price, current_price, atr)
print(f"动态止损价: {stop_price}")
四、实战应用与优化建议
4.1 实战部署流程
步骤1：数据自动化获取
- 使用AkShare等免费数据源
- 设置定时任务（如每天凌晨）更新数据
- 建立数据缓存机制，避免重复请求
步骤2：每日评分计算
- 每天收盘后运行评分模型
- 生成交易信号列表
- 与现有持仓对比
步骤3：交易执行
- 根据信号生成交易计划
- 考虑交易成本（佣金、印花税）
- 执行交易并记录
步骤4：绩效监控
- 每日记录净值
- 每周分析胜率、盈亏比
- 每月优化参数
4.2 常见陷阱与规避方法
陷阱1：过拟合
- 表现：回测收益极高，实盘表现差
- 规避：使用滚动窗口回测、交叉验证、简化模型
陷阱2：数据窥探偏差
- 表现：使用未来信息
- 规避：严格时间序列分割，确保评分计算只使用历史数据
陷阱3：忽略交易成本
- 表现：高频交易导致成本侵蚀利润
- 规避：回测中加入佣金和滑点，设置最低持仓天数
陷阱4：忽视市场变化
- 表现：模型在特定市场有效，环境变化后失效
- 规避：定期重新训练，使用动态权重，设置模型失效预警
4.3 持续优化策略
参数优化
- 使用网格搜索或贝叶斯优化寻找最佳参数
- 限制参数搜索空间，避免过拟合
- 重点关注鲁棒性强的参数组合
策略组合
- 构建多个不同风格的子策略
- 通过相关性分析降低策略间相关性
- 资金分配采用风险平价原则
心理建设
- 坚持模型信号，避免人为干预
- 接受模型的不完美，关注长期表现
- 定期复盘，但不过度频繁调整
五、完整实战案例
以下是一个完整的实战案例，展示从数据获取到交易执行的全过程：
def main():
"""主程序"""
print("=== 股票投资策略打分制量化模型实战 ===")
# 1. 初始化
fetcher = StockDataFetcher()
model = StockScoringModel()
risk_ctrl = RiskController()
# 2. 选择股票池（示例：沪深300成分股中的10只）
symbol_list = ['600519', '000858', '600036', '000333', '601318',
'000651', '600030', '600887', '002415', '000725']
print(f"\n股票池: {symbol_list}")
# 3. 获取数据并计算评分
results = []
sentiment_data = fetcher.get_market_sentiment()
for symbol in symbol_list:
print(f"\n正在分析 {symbol}...")
# 获取数据
price_data = fetcher.get_stock_daily(symbol, '20230101', '20240101')
finance_data = fetcher.get_finance_data(symbol)
if price_data is None or finance_data is None:
continue
# 计算评分
score_result = model.calculate_total_score(symbol, price_data, finance_data, sentiment_data)
results.append(score_result)
print(f" 综合评分: {score_result['total_score']}")
print(f" 基本面: {score_result['fundamental_score']}")
print(f" 技术面: {score_result['technical_score']}")
print(f" 情绪: {score_result['sentiment_score']}")
print(f" 风险: {score_result['risk_score']}")
print(f" 信号: {score_result['signal']}")
# 4. 筛选买入信号
buy_list = [r for r in results if r['signal'] == 'BUY']
print(f"\n=== 买入信号 ({len(buy_list)}只) ===")
for item in sorted(buy_list, key=lambda x: x['total_score'], reverse=True):
print(f"{item['symbol']}: {item['total_score']}分")
# 5. 风险检查
print(f"\n=== 风险检查 ===")
total_capital = 100000
for item in buy_list[:3]: # 只买入前3只
# 模拟买入
price_data = fetcher.get_stock_daily(item['symbol'], '20230101', '20240101')
if price_data is not None:
current_price = price_data['close'].iloc[-1]
position_value = total_capital * 0.1 # 每只10%仓位
# 检查仓位限制
ok, msg = risk_ctrl.check_position_limit(item['symbol'], position_value, total_capital)
print(f"{item['symbol']}: {msg}")
# 检查流动性
avg_volume = price_data['volume'].mean()
current_volume = price_data['volume'].iloc[-1]
ok, msg = risk_ctrl.check_liquidity(item['symbol'], current_volume, avg_volume)
print(f"{item['symbol']}: {msg}")
# 6. 回测验证
print(f"\n=== 回测验证 ===")
backtest = BacktestSystem(initial_capital=100000)
performance = backtest.run_backtest(symbol_list, '20230101', '20240101', model)
if performance:
print("\n回测绩效:")
for k, v in performance.items():
print(f" {k}: {v}")
print("\n=== 实战完成 ===")
if __name__ == "__main__":
main()
六、总结与建议
打分制量化模型是规避情绪化交易、实现长期稳定收益的有效工具。成功的关键在于：
- 坚持纪律：严格按照模型信号执行，避免人为干预
- 持续优化：定期评估模型表现，根据市场变化调整参数
- 风险第一：永远将风险控制放在首位，设置严格的止损纪律
- 长期视角：量化投资是马拉松，关注长期胜率而非单次盈亏
记住，没有完美的模型，只有适合自己的模型。建议从简单的模型开始，逐步增加复杂度，在实践中不断优化，最终形成稳定盈利的交易体系。