7.3量化
-- coding: utf-8 --
import pandas as pd
====================== 全局配置 ======================
BENCHMARK = “000852.XSHG”
INDEX_CODE = “000852.XSHG”
STOCK_NUM = 50
SINGLE_MAX_WEIGHT = 0.03
FACTOR_WEIGHT = {
“pe”: 0.3,
“roe”: 0.3,
“gp”: 0.2,
“mom”: 0.2
}
====================== 因子预处理函数 ======================
def winsorize(ser):
q1 = ser.quantile(0.01)
q99 = ser.quantile(0.99)
return ser.clip(q1, q99)
def standard(ser):
mean_val = ser.mean()
std_val = ser.std()
return (ser - mean_val) / std_val
====================== 初始化 ======================
def initialize(context):
set_benchmark(BENCHMARK)
context.last_month = None
====================== 月度调仓核心 ======================
def rebalance(context):
today = context.current_dt.strftime(“%Y-%m-%d”)
# 获取中证1000成分股并过滤ST、新股
stock_list = get_index_stocks(INDEX_CODE, date=today)
valid = []
for s in stock_list:
info = get_security_info(s)
if “ST” in info.display_name:
continue
start_date = pd.to_datetime(info.start_date)
curr_date = pd.to_datetime(today)
if (curr_date - start_date).days >= 60:
valid.append(s)
target_count = len(valid) if len(valid) < STOCK_NUM else STOCK_NUM # 拉取财务因子 q = query( valuation.code, valuation.pe_ratio, indicator.roe, indicator.gross_profit_margin ).filter(valuation.code.in_(valid)) fund_df = get_fundamentals(q, date=today).set_index("code") fund_df = fund_df.dropna() if len(fund_df) == 0: log.info("无有效财务数据，跳过调仓") return # 计算动量，修复DataFrame取数报错 mom_dict = {} for code in fund_df.index: # 网页免费版get_price 返回DataFrame df = get_price(code, count=20, fields=["close"]) if len(df) < 5: continue # 正确取最后一日/首日收盘价 close_first = df["close"].iloc[0] close_last = df["close"].iloc[-1] mom_dict[code] = close_last / close_first - 1 # 动量缺失兜底 if len(mom_dict) > 0: mom_df = pd.DataFrame(list(mom_dict.items()), columns=["code", "mom"]).set_index("code") all_df = pd.concat([fund_df, mom_df], axis=1).dropna() else: log.info("动量数据缺失，仅使用财务因子打分") all_df = fund_df.copy() all_df["mom"] = 0 if len(all_df) < 1: log.info("无有效因子标的，跳过调仓") return # 多因子加权打分 score = pd.DataFrame(index=all_df.index) score["pe"] = -1 * standard(winsorize(all_df["pe_ratio"])) * FACTOR_WEIGHT["pe"] score["roe"] = standard(winsorize(all_df["roe"])) * FACTOR_WEIGHT["roe"] score["gp"] = standard(winsorize(all_df["gross_profit_margin"])) * FACTOR_WEIGHT["gp"] score["mom"] = standard(winsorize(all_df["mom"])) * FACTOR_WEIGHT["mom"] score["total"] = score.sum(axis=1) target_stocks = score.sort_values("total", ascending=False).head(target_count).index.tolist() single_weight = min(1 / len(target_stocks), SINGLE_MAX_WEIGHT) # 清仓不在目标内的股票 for pos_code in list(context.portfolio.positions.keys()): if pos_code not in target_stocks: order(pos_code, 0) # 买入新持仓，修复行情取数KeyError total_cash = context.portfolio.cash single_cash = total_cash * single_weight for code in target_stocks: df_check = get_price(code, count=1, fields=["close"]) if df_check.empty: continue close_price = df_check["close"].iloc[-1] # A股100股一手 trade_share = int(single_cash / close_price / 100) * 100 if trade_share > 0: order(code, trade_share) log.info(f"月度调仓完成，当期持仓{len(target_stocks)}只股票")每日循环，按月判断调仓
def handle_data(context, data):
cur_month = context.current_dt.month
if cur_month != context.last_month:
rebalance(context)
context.last_month = cur_month