多策略量化分析与改进建议

📅 2026-07-31 📂 obsidian ⏱ 约 98 分钟

date: 2025-08-20 updated: 2025-08-20 conversation_id: a50b27da-a410-458e-9fea-fd5bf7421476 title: "多策略量化分析与改进建议" tags: [deepseek, conversation] ---

多策略量化分析与改进建议

创建时间: 2025-08-20 17:17

👤 用户:

帮我分析这个量化策略，梳理策略的主要逻辑

克隆自聚宽文章：https://www.joinquant.com/post/54907

标题：多策略7.4小改骗点积分

作者：菱形心

from jqdata import * from jqfactor import getfactorvalues import datetime import math from scipy.optimize import minimize

def aftercodechanged(context): unschedule_all()

rundaily(updatelimit_up, "9:05")

# 子策略执行计划 adjusttime = "10:50" checktime = "14:50" if g.portfoliovalueproportion[0] > 0: runweekly(jsgadjust, 1, adjusttime) rundaily(jsgcheck, checktime) if g.portfoliovalueproportion[1] > 0: runmonthly(alldayadjust, 1, adjusttime) if g.portfoliovalueproportion[2] > 0: runmonthly(simpleroaadjust, 1, adjusttime) rundaily(simpleroacheck, checktime) if g.portfoliovalueproportion[3] > 0: runmonthly(weakcycadjust, 1, adjusttime) if g.portfoliovalueproportion[4] > 0: runweekly(strategyadjust, 1, adjusttime) # 调整为10:00 rundaily(strategycheck, checktime) # 保持涨停检查时间 # 每日剩余资金购买货币ETF

rundaily(endtrade, "14:55") log.info("策略代码改变")

初始化函数，设定基准等等

def initialize(context): # 设定沪深300作为基准 # setbenchmark("515080.XSHG") # 打开防未来函数 setoption("avoidfuturedata", True) # 开启动态复权模式(真实价格) setoption("userealprice", True) # 输出内容到日志 log.info() log.info("初始函数开始运行且全局只运行一次") # 过滤掉order系列API产生的比error级别低的log log.setlevel("order", "error") # 固定滑点设置ETF 0.001(即交易对手方一档价) setslippage(FixedSlippage(0.002), type="fund") # 股票交易总成本0.3%(含固定滑点0.02) setslippage(FixedSlippage(0.02), type="stock") setordercost( OrderCost( opentax=0, closetax=0.001, opencommission=0.0003, closecommission=0.0003, closetodaycommission=0, mincommission=5, ), type="stock", ) # 设置货币ETF交易佣金0 setordercost( OrderCost( opentax=0, closetax=0, opencommission=0, closecommission=0, closetodaycommission=0, mincommission=0, ), type="mmf", ) # 全局变量 g.fillstock = "511880.XSHG" # 货币ETF,用于现金管理 g.strategys = {} g.portfoliovalueproportion = [0, 0.3, 0.1, 0.2, 0.4] g.positions = {i: {} for i in range(len(g.portfoliovalue_proportion))} # 记录每个子策略的持仓股票

rundaily(updatelimit_up, "9:05")

# 子策略执行计划 adjusttime = "14:35" checktime = "14:50" if g.portfoliovalueproportion[0] > 0: runweekly(jsgadjust, 1, adjusttime) rundaily(jsgcheck, checktime) if g.portfoliovalueproportion[1] > 0: runmonthly(alldayadjust, 1, adjusttime) if g.portfoliovalueproportion[2] > 0: runmonthly(simpleroaadjust, 1, adjusttime) rundaily(simpleroacheck, checktime) if g.portfoliovalueproportion[3] > 0: runmonthly(weakcycadjust, 1, adjusttime) if g.portfoliovalueproportion[4] > 0: runweekly(strategyadjust, 1, adjusttime) # 调整为10:00 rundaily(strategycheck, checktime) # 保持涨停检查时间 # 每日剩余资金购买货币ETF

rundaily(endtrade, "14:55")

def processinitialize(context): print("重启程序") g.strategys["搅屎棍策略"] = JSGStrategy(context, index=0, name="搅屎棍策略") g.strategys["全天候策略"] = AllDayStrategy(context, index=1, name="全天候策略") g.strategys["简单ROA策略"] = SimpleROAStrategy(context, index=2, name="简单ROA策略") g.strategys["弱周期价投策略"] = WeakCycStrategy(context, index=3, name="弱周期价投策略") g.strategys["小市值策略"] = Strategy_SmallCap(context, index=4, name="小市值策略")

def saveallpositions(context): """保存所有策略的持仓信息到数据库""" if context.runparams.type != 'simtrade': return positions = [] for code, pos in context.portfolio.positions.items(): positions.append({ 'code': pos.security, 'volume': pos.totalamount, 'cost': pos.avgcost }) try: g.jqqmtapi.update_positions("多策略V1.0", positions) log.info("所有策略持仓信息已保存到数据库") except Exception as e: log.error(f"策略持仓更新失败:{str(e)}")

def updatelimitup(context): for s in g.strategys.values(): s.updateyesterdaylimit_up()

def onevent(context, event): for s in g.strategys.values(): s.onevent(event)

def jsg_check(context): g.strategys["搅屎棍策略"].check()

def jsg_adjust(context): g.strategys["搅屎棍策略"].adjust()

def alldayadjust(context): g.strategys["全天候策略"].adjust()

def simpleroaadjust(context): g.strategys["简单ROA策略"].adjust()

def simpleroacheck(context): g.strategys["简单ROA策略"].check()

def weakcycadjust(context): g.strategys["弱周期价投策略"].adjust()

def strategy_check(context): g.strategys["小市值策略"].check()

def strategy_adjust(context): g.strategys["小市值策略"].adjust()

买入货币ETF

尾盘处理

def endtrade(context): currentdata = getcurrentdata()

# 卖出未记录的股票（比如送股） keys = [key for d in g.positions.values() if isinstance(d, dict) for key in d.keys()] for stock in context.portfolio.positions: if stock not in keys and stock != g.fillstock and currentdata[stock].lastprice < currentdata[stock].highlimit: if ordertarget_value(stock, 0): log.info(f"卖出{stock}因送股未记录在持仓中")

# 买入货币ETF # amount = int(context.portfolio.availablecash / currentdata[g.fillstock].lastprice) # if amount >= 100: # order(g.fill_stock, amount)

卖出货币ETF换现金

def getcash(context, value): if g.fillstock not in context.portfolio.positions: return currentdata = getcurrentdata() amount = math.ceil(value / currentdata[g.fillstock].lastprice / 100) * 100 position = context.portfolio.positions[g.fillstock].closeableamount if amount >= 100: order(g.fill_stock, -min(amount, position))

策略基类

class Strategy:

def init(self, context, index, name): self.context = context self.index = index self.name = name self.stocksum = 1 self.holdlist = [] self.min_volume = 2000

# 增加过滤参数配置 self.filterst = True # 是否过滤ST股票 self.filtercyb = False # 是否过滤创业板 self.filterkcb = True # 是否过滤科创板 self.filterbjb = True # 是否过滤北交所 self.filterthird = True # 是否过滤三板 self.newstock_days = 375 # 次新股过滤的天数间隔

# 增加涨跌停交易配置 self.allowlimitupsell = False # 是否允许涨停卖出 self.allowlimitdownbuy = False # 是否允许跌停买入

# 增加开盘涨跌停过滤开关 self.filteropenlimit = False # 是否过滤开盘涨跌停

# 增加资金分配配置 self.yesterdaylimitup = [] # 添加昨日涨停股票列表属性 # 注册分红送股事件处理 self.context.handle_dividend = True # 打开分红送股事件监听

self.limitstocksumbeforesell = True # 卖出前是否限制stock_sum只股票

# 增加清仓记录相关配置 self.soldstockrecord = {} # 记录卖出股票的时间 self.exclude_days = 20 # 卖出后多少天不允许买入

def filtersoldstock(self, stocklist): """过滤最近卖出的股票""" currentdate = self.context.currentdt.date() return [stock for stock in stocklist if stock not in self.soldstockrecord or (currentdate - self.soldstockrecord[stock]).days >=self.excludedays]

def recordsoldstock(self, stock): """记录卖出股票""" # 移除过期的记录 currentdate = self.context.currentdt.date() expiredstocks = [s for s, date in self.soldstockrecord.items() if (currentdate - date).days > self.excludedays] for s in expiredstocks: self.soldstockrecord.pop(s)

# 记录新的卖出 self.soldstockrecord[stock] = current_date

def onevent(self, event): """处理分红送股事件""" if event.name == 'Dividends': for dividend in event.dividends: if "scalefactor" not in dividend: continue # 获取分红送股信息 stock = event.security.code if stock in g.positions[self.index]: # 更新持仓数量为实际持仓数量 preposition = g.positions[self.index][stock] realposition = round(preposition * dividend["scalefactor"], 0) if realposition > 0: g.positions[self.index][stock] = realposition else: g.positions[self.index].pop(stock, None) log.info(f"[{self.name}] 更新{stock}送股后持仓为: {realposition}, 之前为{preposition},scalefactor:{dividend['scalefactor']}")

# 获取策略当前持仓市值 def gettotalvalue(self): if not g.positions[self.index]: return 0 return sum(self.context.portfolio.positions[key].price * value for key, value in g.positions[self.index].items())

def updateyesterdaylimitup(self): """更新昨日涨停股票列表""" self.yesterdaylimit_up = []

if self.holdlist: dfyesterday = getprice( self.holdlist, enddate=self.context.previousdate, frequency='daily', fields=['close', 'highlimit'], count=1, panel=False, fillpaused=False ) self.yesterdaylimitup = list(dfyesterday[dfyesterday['close'] == dfyesterday['highlimit']].code)

# 检查昨日涨停票 def check(self, stocks=None): if not stocks: stocks = self.holdlist

checkstocks = [stock for stock in stocks if stock in self.yesterdaylimitup] if not checkstocks: return []

# 获取当前价格数据 currentdata = getcurrent_data()

# 返回昨日涨停今日未涨停的股票 return [stock for stock in checkstocks if currentdata[stock].lastprice < currentdata[stock].high_limit]

def closeposition(self, stock): if self.ordertargetvalue(stock, 0): self.recordsoldstock(stock) return True else: return False

# 调仓(等权购买target中按顺序排列固定数量的的标的) def adjust(self, target): # 获取前stocksum个标的 if self.limitstocksumbeforesell: target = target[: min(len(target), self.stock_sum)]

# 获取已持有列表 portfolio = self.context.portfolio

# 调仓卖出 for stock in self.holdlist: if stock not in target: self.closeposition(stock)

if self.limitstocksumbeforesell: # 调仓买入 count = len(set(target) - set(self.holdlist)) if count == 0 or self.stocksum <= len(self.holdlist): return else: availablepositions = max(0, self.stocksum - len(self.holdlist)) if availablepositions == 0: return tobuyset = set(target) - set(self.holdlist) target = [s for s in target if s in tobuyset][:availablepositions] count = len(target) # 目标市值 targetvalue = portfolio.totalvalue * g.portfoliovalue_proportion[self.index]

# 当前市值 positionvalue = self.gettotal_value()

# 可用现金:当前现金 + 货币ETF市值 availablecash = portfolio.availablecash + (portfolio.positions[g.fillstock].value if g.fillstock in portfolio.positions else 0)

# 买入股票的总市值 value = max(0, min(targetvalue - positionvalue, available_cash))

# 卖出部分货币ETF获取现金 if value > portfolio.availablecash: getcash(self.context, value - portfolio.available_cash)

# 等价值买入每一个未买入的标的 for security in target: if security not in self.holdlist: self.ordertargetvalue(security, value / count)

# 调仓2(targets为字典，key为股票代码，value为目标市值) def _adjust2(self, targets):

# 获取已持有列表 currentdata = getcurrent_data() portfolio = self.context.portfolio

# 清仓被调出的 for stock in self.holdlist: if stock not in targets: self.closeposition(stock)

# 先卖出 for stock, target in targets.items(): price = currentdata[stock].lastprice value = g.positions[self.index].get(stock, 0) price if value - target > self.minvolume and value - target > price 100: self.ordertargetvalue(stock, target)

# 后买入 for stock, target in targets.items(): price = currentdata[stock].lastprice value = g.positions[self.index].get(stock, 0) price if target - value > self.minvolume and target - value > price 100: if target - value > portfolio.availablecash: getcash(self.context, target - value - portfolio.availablecash) if portfolio.availablecash > price * 100: self.ordertargetvalue(stock, target)

# 自定义下单(涨跌停不交易) def ordertargetvalue(self, security, value): currentdata = getcurrentdata()

# 检查标的是否停牌 if current_data[security].paused: log.info(f"{security}: 今日停牌") return False

# 获取当前标的的价格 price = currentdata[security].lastprice

# 获取当前策略的持仓数量 current_position = g.positions[self.index].get(security, 0)

# 计算目标持仓数量 target_position = (int(value / price) // 100) * 100 if price != 0 else 0

# 计算需要调整的数量 adjustment = targetposition - currentposition

# 涨跌停检查逻辑优化 if adjustment > 0: # 买入 if currentdata[security].lastprice == currentdata[security].lowlimit: if not self.allowlimitdownbuy: log.info(f"{security}: 当前跌停，不允许买入") return False elif currentdata[security].lastprice == currentdata[security].highlimit: log.info(f"{security}: 当前涨停，无法买入") return False else: # 卖出 if currentdata[security].lastprice == currentdata[security].highlimit: if not self.allowlimitupsell: log.info(f"{security}: 当前涨停，不允许卖出") return False elif currentdata[security].lastprice == currentdata[security].lowlimit: log.info(f"{security}: 当前跌停，无法卖出") return False

# 检查是否当天买入卖出 closeableamount = self.context.portfolio.positions[security].closeableamount if security in self.context.portfolio.positions else 0 if adjustment < 0 and closeable_amount == 0: log.info(f"{security}: 当天买入不可卖出") return False

# 下单并更新持仓 if adjustment != 0: o = order(security, adjustment) if o: # 记录交易日志 action = "买入" if o.is_buy else "卖出" #log.info(f"[{self.name}] {action} {security}: {o.amount}股, 价格: {price:.2f}")

# 更新持仓数量 g.positions[self.index][security] = (o.isbuy and 1 or -1) * o.amount + currentposition # 如果目标持仓为零，移除该证券 if targetposition == 0: g.positions[self.index].pop(security, None) # 更新持有列表 self.holdlist = list(g.positions[self.index].keys()) return True else: return False

# 基础过滤(过滤科创北交、ST、停牌、次新股) def filterbasicstock(self, stock_list):

currentdata = getcurrentdata() filteredstocks = []

for stock in stocklist: if currentdata[stock].paused: continue

# 开盘涨跌停过滤 if self.filteropenlimit and ( currentdata[stock].dayopen == currentdata[stock].highlimit or currentdata[stock].dayopen == currentdata[stock].lowlimit ): continue

# ST股票过滤 if self.filterst and (currentdata[stock].isst or "ST" in currentdata[stock].name or "*" in currentdata[stock].name or "退" in currentdata[stock].name): continue

# 板块过滤 if ((self.filtercyb and stock.startswith('300')) or (self.filterkcb and stock.startswith('688')) or (self.filterbjb and stock.startswith('8')) or (self.filterthird and stock.startswith('4'))): continue

# 次新股过滤 if self.newstockdays > 0 and ( self.context.previousdate - getsecurityinfo(stock).startdate < datetime.timedelta(self.newstockdays)): continue

filtered_stocks.append(stock)

return filtered_stocks

# 过滤当前时间涨跌停的股票 def filterlimituplimitdownstock(self, stocklist, upratio=1, downratio=1): currentdata = getcurrentdata() return [ stock for stock in stocklist if currentdata[stock].lastprice < currentdata[stock].highlimit upratio and currentdata[stock].lastprice > currentdata[stock].lowlimit downratio ] currentdata = getcurrentdata() return [ stock for stock in stocklist if currentdata[stock].lastprice < currentdata[stock].highlimit and currentdata[stock].lastprice > currentdata[stock].lowlimit ]

# 判断今天是在空仓月 def isemptymonth(self): month = self.context.currentdt.month return month in self.passmonths

搅屎棍策略

class JSG_Strategy(Strategy):

def init(self, context, index, name): super().init(context, index, name)

self.stocksum = 6 # 判断买卖点的行业数量 self.num = 1 # 空仓的月份 self.passmonths = [1, 4]

def getStockIndustry(self, stocks): industry = getindustry(stocks) return pd.Series({stock: info["swl1"]["industryname"] for stock, info in industry.items() if "swl1" in info})

# 获取市场宽度 def getmarketbreadth(self): # 指定日期防止未来数据 yesterday = self.context.previousdate # 获取初始列表 stocks = getindexstocks("000985.XSHG") count = 1 h = getprice( stocks, enddate=yesterday, frequency="1d", fields=["close"], count=count + 20, panel=False, ) h["date"] = pd.DatetimeIndex(h.time).date dfclose = h.pivot(index="code", columns="date", values="close").dropna(axis=0) # 计算20日均线 dfma20 = dfclose.rolling(window=20, axis=1).mean().iloc[:, -count:] # 计算偏离程度 dfbias = dfclose.iloc[:, -count:] > dfma20 dfbias["industryname"] = self.getStockIndustry(stocks) # 计算行业偏离比例 dfratio = ((dfbias.groupby("industryname").sum() * 100.0) / dfbias.groupby("industryname").count()).round() # 获取偏离程度最高的行业 topvalues = dfratio.loc[:, yesterday].nlargest(self.num) I = top_values.index.tolist() return I

# 过滤股票 def filter(self): stocks = getindexstocks("399101.XSHE") # stocks = getallsecurities("stock", date=self.context.previousdate).index.tolist() stocks = self.filterbasicstock(stocks) stocks = ( getfundamentals( query( valuation.code, ) .filter( valuation.code.in(stocks), indicator.adjustedprofit > 0, ) .orderby(valuation.marketcap.asc()) ) .head(20) .code ) stocks = self.filterlimituplimitdown_stock(stocks) return stocks

# 择时 def select(self): I = self.getmarketbreadth() industries = {"银行I", "煤炭I", "采掘I", "钢铁I"} if not industries.intersection(I) and not self.isemptymonth(): return self.filter() return []

# 调仓 def adjust(self): self._adjust(self.select())

# 获取昨日涨停票 def check(self): bannerstocks = self.check() for stock in bannerstocks: self.ordertargetvalue(stock, 0)

全天候ETF策略

class AllDayStrategy(Strategy):

def init(self, context, index, name): super().init(context, index, name)

# 最小交易额(限制手续费) self.minvolume = 2000 # 全天候ETF组合参数 self.etfpool = [ "511260.XSHG", # 十年国债ETF "518880.XSHG", # 黄金ETF "513100.XSHG", # 纳指100 # 2020年之后成立(注意回测时间) "159980.XSHE", # 有色ETF "162411.XSHE", # 华宝油气LOF "159985.XSHE", # 豆粕ETF ] # 标的仓位占比 self.rates = [0.45, 0.25, 0.15, 0.05, 0.05, 0.05]

# 调仓 def adjust(self): totalvalue = self.context.portfolio.totalvalue g.portfoliovalueproportion[self.index] # 计算每个 ETF 的目标价值 targets = {etf: totalvalue rate for etf, rate in zip(self.etfpool, self.rates)} self._adjust2(targets)

简单ROA策略

class SimpleROAStrategy(Strategy): def init(self, context, index, name): super().init(context, index, name)

self.stock_sum = 1

def filter(self): stocks = getallsecurities("stock", date=self.context.previousdate).index.tolist() stocks = self.filterbasicstock(stocks) stocks = list( getfundamentals( query(valuation.code, indicator.roa).filter( valuation.code.in(stocks), valuation.pbratio > 0, valuation.pbratio < 1, cashflow.subtotaloperatecashinflow > 1e6, indicator.adjustedprofit > 1e6, indicator.increturn > 0, indicator.incnetprofityearonyear > -13, cashflow.subtotaloperatecashinflow/indicator.adjustedprofit > 1, cashflow.netoperatecashflow > 0, ) ) .sortvalues(by="roa", ascending=False) .head(10) .code ) stocks = self.filterlimituplimitdown_stock(stocks) return stocks

# 调仓 def adjust(self): self.adjust(self.filter()[:self.stocksum])

# 检查昨日涨停票 def check(self): bannerstocks = self.check() if bannerstocks: target = [stock for stock in self.filter() if stock not in bannerstocks] self._adjust(target)

def calculatedividendratio(stocklist, enddate, bonusyears=1, bonusdatefield='boardplanpubdate', dropna=False): # 获取市值数据 q = query(valuation.code, valuation.marketcap).filter( valuation.code.in(stocklist) ) marketcapdf = getfundamentals(q, date=enddate - datetime.timedelta(days=1)) marketcapdf.set_index('code', inplace=True)

# 初始化分红DataFrame,添加bonus列并初始化为0 bonusdf = pd.DataFrame(index=stocklist) bonus_df['bonus'] = 0

batchsize = 1000 // bonusyears startdate = enddate - datetime.timedelta(days=365 * bonus_years)

# 批量获取分红数据 for i in range(0, len(stocklist), batchsize): batchstocks = stocklist[i:i+batchsize] q = query(finance.STKXRXD).filter( finance.STKXRXD.code.in(batchstocks), getattr(finance.STKXRXD, bonusdatefield) >= startdate, getattr(finance.STKXRXD, bonusdatefield) < enddate ) tmpdf = finance.runquery(q) if not tmpdf.empty: grouped = tmpdf.groupby('code')['bonusamountrmb'].sum() bonusdf.loc[grouped.index, 'bonus'] = grouped

# 处理NA值 if dropna: bonusdf = bonusdf[bonusdf['bonus'].notna()] else: bonus_df.fillna(0, inplace=True)

# 计算年化股息率 bonusdf['dividendratio'] = bonusdf['bonus'] / bonusyears / (marketcapdf['marketcap'] * 1e4) bonusdf['dividendratio'] = bonusdf['dividend_ratio'].fillna(0)

return bonus_df

弱周期价投策略

class WeakCycStrategy(Strategy): def init(self, context, index, name): super().init(context, index, name)

self.stocksum = 5 self.bondetf = "511260.XSHG" # 最小交易额(限制手续费) self.minvolume = 2000 self.pemean = 20 self.time = 0

self.init_stocks = [ "601985.XSHG", # 中国核电 "600025.XSHG", # 华能水电 "600900.XSHG", # 长江电力 "003816.XSHE", # 中国广核 "600886.XSHG", # 国投电力 "600905.XSHG", # 三峡能源

"000429.XSHE", # 粤高速 "600377.XSHG", # 宁沪高速 "600269.XSHG", # 赣粤高速 "600033.XSHG", # 福建高速 "600035.XSHG", # 楚天高速

"600941.XSHG", # 中国电信 "601728.XSHG", # 中国移动

"601088.XSHG", # 中国神华 "601225.XSHG", # 陕西煤业 "600938.XSHG", # 中海油 "601857.XSHG", # 中石油 "601899.XSHG", # 紫金矿业 ]

def getstocksum(self):

if self.pemean < 20: self.stocksum = 4 elif self.pemean < 30: self.stocksum = 2 else: self.stock_sum = 0

def select(self): yesterday = self.context.previousdate stocks = self.initstocks data = getfundamentals(query(valuation.code, valuation.peratio, valuation.marketcap).filter(valuation.code.in(stocks))) totalmarketcap = data.marketcap.sum() self.pemean = totalmarketcap / (1 / data.peratio * data.marketcap).sum() self.getstocksum()

# 获取基本面筛选的股票 stocks = getfundamentals( query(valuation.code) .filter( valuation.code.in(stocks), valuation.marketcap > 200, valuation.peratio < 20, indicator.adjustedprofit > 0, indicator.incnetprofityearonyear > 1, indicator.grossprofitmargin > 20, # 毛利 ) ).code

# 计算股息率并筛选大于3%的股票 bonusdf = calculatedividendratio(list(stocks), self.context.currentdt, bonusyears=1) bonusdf = bonusdf[bonusdf['dividendratio'] > 0.03] bonusdf = bonusdf.sortvalues('dividendratio', ascending=False) stocks = list(bonusdf.index)

return stocks

def adjust(self): stocks = self.select()[: self.stocksum] # self.adjust(stocks) stocks.append(self.bondetf) rates = [round(1 / (self.stocksum + 2), 3)] (len(stocks) - 1) rates.append(round(1 - sum(rates), 3)) totalvalue = self.context.portfolio.totalvalue g.portfoliovalueproportion[self.index] targets = {stock: total_value * rate for stock, rate in zip(stocks, rates)}

self._adjust2(targets)

7年40倍策略

class Strategy_7x40(Strategy):

def init(self, context, index, name): super().init(context, index, name) self.filtercyb = False # 是否过滤创业板 # 判断买卖点的行业数量 self.num = 1 # 策略参数设置 self.stocksum = 50 # 持股数量 self.pass_months = [1,4] # 空仓月份 self.weights = [1.0, 1.0, 1.6, 0.8, 2.0] # 多因子权重

def getStockIndustry(self, stocks): industry = getindustry(stocks) return pd.Series({stock: info["swl1"]["industryname"] for stock, info in industry.items() if "swl1" in info})

# 获取市场宽度 def getmarketbreadth(self): # 指定日期防止未来数据 yesterday = self.context.previousdate # 获取初始列表 stocks = getindexstocks("000985.XSHG") count = 1 h = getprice( stocks, enddate=yesterday, frequency="1d", fields=["close"], count=count + 20, panel=False, ) h["date"] = pd.DatetimeIndex(h.time).date dfclose = h.pivot(index="code", columns="date", values="close").dropna(axis=0) # 计算20日均线 dfma20 = dfclose.rolling(window=20, axis=1).mean().iloc[:, -count:] # 计算偏离程度 dfbias = dfclose.iloc[:, -count:] > dfma20 dfbias["industryname"] = self.getStockIndustry(stocks) # 计算行业偏离比例 dfratio = ((dfbias.groupby("industryname").sum() * 100.0) / dfbias.groupby("industryname").count()).round() # 获取偏离程度最高的行业 topvalues = dfratio.loc[:, yesterday].nlargest(self.num) I = top_values.index.tolist() return I

# 获取单只股票N个单位时间前的收盘价 def getclose(self, stock, n, unit): return attributehistory(stock, n, unit, 'close')['close'][0]

# 获取现价相对N个单位前价格的涨幅 def getreturn(self, stock, n, unit): pricebefore = attributehistory(stock, n, unit, 'close')['close'][0] pricenow = self.getclose(stock, 1, '1m') if not isnan(pricenow) and not isnan(pricebefore) and pricebefore != 0: return pricenow / pricebefore else: return 100

# 选股主逻辑 def select(self): # 获取初始股票池 yesterday = self.context.previousdate initiallist = getallsecurities('stock', yesterday).index.tolist() initiallist = self.filterbasicstock(initiallist)

# 获取财务指标筛选 q = query( valuation.code, valuation.marketcap, valuation.circulatingmarketcap ).filter( valuation.code.in(initiallist), valuation.pbratio > 0, indicator.increturn > 0, indicator.inctotalrevenueyearonyear > 0, indicator.incnetprofityearonyear > 0 ).orderby( valuation.market_cap.asc() ).limit(100)

df = getfundamentals(q, date=yesterday) df.index = df.code initiallist = list(df.index)

# 获取因子原始值 MC, CMC, PN, TV, RE = [], [], [], [], [] for stock in initiallist: mc = df.loc[stock]['marketcap'] MC.append(mc) cmc = df.loc[stock]['circulatingmarketcap'] CMC.append(cmc) pricenow = self.getclose(stock, 1, '1m') PN.append(pricenow) totalvolumen = attributehistory(stock, 1200, '1m', 'volume')['volume'].sum() TV.append(totalvolumen) mdaysreturn = self.getreturn(stock, 60, '1d') RE.append(mdays_return)

# 合并数据 df = pd.DataFrame(index=initiallist, columns=['marketcap','circulatingmarketcap','pricenow','totalvolumen','mdaysreturn']) df['marketcap'] = MC df['circulatingmarketcap'] = CMC df['pricenow'] = PN df['totalvolumen'] = TV df['mdays_return'] = RE df = df.dropna()

# 计算因子得分 min0, min1, min2, min3, min4 = min(MC), min(CMC), min(PN), min(TV), min(RE) templist = [] for i in range(len(list(df.index))): score = (self.weights[0] math.log(min0 / df.iloc[i,0]) + self.weights[1] math.log(min1 / df.iloc[i,1]) + self.weights[2] math.log(min2 / df.iloc[i,2]) + self.weights[3] math.log(min3 / df.iloc[i,3]) + self.weights[4] * math.log(min4 / df.iloc[i,4])) templist.append(score) df['score'] = temp_list

# 排序并返回最终选股列表 df = df.sortvalues(by='score', ascending=False) finallist = list(df.index)

# 过滤涨跌停股票 finallist = self.filterlimituplimitdownstock(final_list)

return final_list

def filter(self): I = self.getmarketbreadth() industries = {"银行I", "煤炭I", "采掘I", "钢铁I"} if not industries.intersection(I) and not self.isemptymonth(): return self.select() return [] # 调仓 def adjust(self): self._adjust(self.filter())

# 检查持仓 def check(self): bannerstocks = self.check() for stock in bannerstocks: self.ordertargetvalue(stock, 0)

class StrategySmallCap(Strategy): def init(self, context, index, name): super().init(context, index, name) # 策略参数设置 self.stocksum = 9 # 持股数量 self.passmonths = [1, 4] # 空仓月份 self.stoplosslimit = 0.1 # 个股止损线 self.stoplossmarket = 0.05 # 市场趋势止损参数 self.filtercyb = True # 过滤创业板 self.filterkcb = True # 过滤科创板 self.filterbjb = True # 过滤北交所 self.newstockdays = 375 # 次新股过滤天数

def check(self): """每日14:00执行涨停检查和补仓""" if not self.isemptymonth(): # 检查昨日涨停股票是否打开，使用父类的check方法 bannerstocks = self.check() for stock in bannerstocks: self.ordertargetvalue_(stock, 0) log.info(f"[{self.name}] {stock}涨停打开，卖出")

def select(self): if self.isemptymonth(): return [] # 获取初始股票池 initiallist = getindex_stocks('399101.XSHE') # 中小板综指成分股

filterlist = self.filterbasicstock(initiallist) # 分开处理持仓和非持仓股票 holdstocks = list(g.positions[self.index].keys()) nonholdstocks = [stock for stock in initiallist if stock not in hold_stocks]

# 基础过滤 filterlist = self.filterbasicstock(initiallist)

# 只对非持仓股票进行涨跌停过滤 nonholdfilter = [stock for stock in filterlist if stock not in holdstocks] nonholdfilter = self.filterlimituplimitdownstock(nonhold_filter)

# 合并持仓股票 holdfilter = [stock for stock in filterlist if stock in holdstocks] filterlist = nonholdfilter + holdfilter # 获取财务指标筛选 q = query( valuation.code, valuation.marketcap ).filter( valuation.code.in(filterlist), valuation.marketcap.between(5, 300) # 市值区间5-300亿 ).orderby( valuation.market_cap.asc() )

df = getfundamentals(q) df = df.sortvalues('marketcap', ascending=True) finallist = list(df.code)[:self.stock_sum]

return final_list

def adjust(self): targetlist = self.select() self.adjust(target_list)

💭 思考过程

嗯，让我仔细看看这个量化策略代码。这看起来是一个多策略组合，包含了好几个不同的子策略。用户想要我梳理主要逻辑，需要深入理解每个部分。

先看看整体结构，这是一个在聚宽平台上运行的量化交易策略，用了Python写的。初始化部分设置了各种参数，比如交易成本、滑点什么的，还定义了几个子策略。

哦，这里有个全局变量g.portfoliovalueproportion，用来分配资金到不同策略的，比例是[0, 0.3, 0.1, 0.2, 0.4]。看起来第一个策略现在是0，可能被禁用了。

策略基类Strategy定义了很多通用功能，比如过滤ST股票、处理涨跌停、记录卖出股票等。然后有五个具体的策略类继承了这个基类。

搅屎棍策略这个名字有点意思……它好像是基于市场宽度来选择行业的，避开银行、煤炭这些周期性强的行业，然后选小市值的股票。

全天候策略就比较简单了，就是投资几个不同的ETF，国债、黄金、纳斯达克什么的，按固定比例分配。

简单ROA策略是选ROA高、PB小于1的股票，看起来是价值投资思路。

弱周期价投策略专门投资那些弱周期的高股息股票，像电力、高速路、电信这些，还会根据市场PE调整仓位。

小市值策略就是纯粹选小市值的股票，从中小板里选市值最小的。

这些策略都有自己的调仓时间，有的是每周，有的是每月。每天还会检查有没有昨天涨停今天没涨停的股票，有的话就卖掉。

整体来看，这个策略组合试图通过不同策略之间的低相关性来降低风险，同时抓住不同市场环境下的机会。每个子策略都有自己的逻辑和偏好，组合起来应该比单一策略更稳健。

我觉得可以给用户解释每个策略的核心逻辑，以及它们是如何组合在一起的，还有整个策略的风控措施和资金管理方法。这样用户就能理解这个量化策略是怎么运作的了。

好的，我来为您详细梳理和分析这个名为“多策略7.4小改”的量化策略。

策略核心逻辑总览

这是一个多策略组合模型。其核心思想是将总资金按固定比例分配给多个相互独立或低相关性的子策略，由一个大类资产（货币ETF 511880）进行现金管理。每个子策略专注于不同的市场风格（如小市值、高股息、动量反转等），旨在通过策略分散化来平滑整体收益曲线，降低单一策略失效的风险，追求更稳健的复合回报。

策略框架与资金分配

策略将投资组合分为5个部分，由一个主引擎调度：

  1. 资金分配 (g.portfoliovalueproportion): [0, 0.3, 0.1, 0.2, 0.4]
  • 索引0 (0%): “搅屎棍策略” (当前未启用)
  • 索引1 (30%): “全天候策略” (ETF资产配置)
  • 索引2 (10%): “简单ROA策略” (深度价值)
  • 索引3 (20%): “弱周期价投策略” (高股息防御)
  • 索引4 (40%): “小市值策略” (小盘股风格)
  1. 持仓管理 (g.positions): 一个字典，按索引记录每个子策略分别持有哪些股票以及数量，实现了策略间的持仓隔离。
  1. 运行调度: 策略根据每个子策略的资金比例是否大于0来动态安排其运行计划（runmonthly, runweekly, run_daily），主要在每个交易日的 14:35 (调整) 和 14:50 (检查) 执行。
  1. 现金管理: 几乎所有的闲置现金都会在尾盘 (14:55) 买入货币ETF 511880.XSHG 来增厚收益。当子策略需要现金时，会先卖出部分货币ETF。

各子策略逻辑详解

1. 小市值策略 (Strategy_SmallCap) - 主力策略 (40%)

  • 核心逻辑: 相信“小盘股效应”，即长期来看小市值公司的股票回报率优于大市值公司。
  • 选股方法:
  1. 股票池: 以中小板综指 (399101.XSHE) 的成分股为初选池。
  2. 过滤: 应用基类的基础过滤（剔除ST、次新、停牌、科创板/创业板/北交所股票等）。
  3. 核心指标: 在过滤后的股票中，选择总市值最小的前9只股票。
  • 调仓: 每月调一次仓，卖出不在新名单中的股票，等权重买入新名单中的股票。
  • 风控: 在1月和4月空仓，以规避年报和季报披露期小盘股可能出现的风险。

2. 全天候策略 (AllDayStrategy) - 防御策略 (30%)

  • 核心逻辑: 模仿桥水的“全天候”理念，通过配置低相关性的各类资产来抵御不同经济周期的风险，追求稳健。
  • 投资标的: 完全投资于ETF和LOF基金，不直接投资股票。
  • 511260 (10年期国债ETF) - 利率风险
  • 518880 (黄金ETF) - 通胀风险
  • 513100 (纳指ETF) - 海外成长股风险
  • 159980 (有色ETF), 162411 (华宝油气), 159985 (豆粕ETF) - 商品风险
  • 调仓: 每月调一次仓，将资金按固定比例([0.45, 0.25, 0.15, 0.05, 0.05, 0.05])分配到这些ETF上。

3. 弱周期价投策略 (WeakCycStrategy) - 稳健策略 (20%)

  • 核心逻辑: 投资于受经济周期影响小、盈利能力稳定、高分红的企业，追求股息收益和稳健增长。
  • 选股方法:
  1. 初选池: 一个预设的股票列表，包含核电、水电、高速、通信、煤炭等防御性行业龙头。
  2. 财务过滤: 要求PE < 20、利润正增长、毛利率 > 20%。
  3. 核心指标: 计算过去1年的股息率，并筛选出股息率 > 3% 的股票。
  • 动态仓位: 计算一篮子防御股的平均PE，用来动态决定股票仓位。市场越便宜（PE低），股票仓位越高；反之则降低股票仓位，增加债券ETF(511260)的配置。

4. 简单ROA策略 (SimpleROAStrategy) - 深度价值策略 (10%)

  • 核心逻辑: 寻找“便宜的好公司”，即估值低（破净）但盈利能力强的企业。
  • 选股方法:
  1. 财务筛选:
  • PB < 1 (破净)
  • ROA (资产收益率) 高 - 按此排序
  • 现金流充沛 (经营现金流入 > 利润)
  • 利润正增长
  • 持仓: 通常只持有1只筛选出的最佳股票，集中投资。
  • 风控: 每日检查，如果持有的股票昨日涨停但今日未涨停，会将其卖出并换入备选股中下一个符合条件的股票。

5. 搅屎棍策略 (JSG_Strategy) - 均值回归策略 (0%，当前未启用)

  • 核心逻辑: 基于市场宽度进行择时。当多数股票跌破20日线，市场情绪极度悲观时，买入超跌的小市值股票，博弈市场反弹。
  • 择时信号: 计算中证全指成分股中股价高于20日线的比例。只有当没有出现银行、煤炭等强周期行业领涨的“虚假繁荣”时，且不在空仓月，才会开仓。
  • 选股: 选择市值最小的20只股票。

通用功能与风控措施 (策略基类 Strategy)

这是策略非常扎实的一部分，体现了丰富的实战经验：

  1. 股票过滤器 (filterbasicstock): 自动过滤ST、停牌、次新、科创板、创业板、北交所股票，是A股量化策略的标配。
  2. 涨跌停处理 (ordertargetvalue_): 下单前检查涨跌停状态，默认情况下不在涨停时卖出（惜售）也不在跌停时买入（避免流动性陷阱），但行为可通过参数调整。
  3. 昨日涨停检查 (_check): 一个重要的止盈/风控机制。如果持仓股昨日涨停但今日未封板，意味着短期动量可能衰竭，策略会选择卖出。
  4. 卖出冷却期 (soldstockrecord): 卖出某只股票后，将其放入“冷宫”（默认20天内不再买入），避免反复打脸。
  5. 分红送股处理 (on_event): 监听分红送股事件，并自动调整持仓记录中的数量，确保账户记录准确。

策略优点

  1. 真正的多策略分散: 不是多个因子，而是多个逻辑迥异的策略，相关性低，能有效降低整体回撤。
  2. 逻辑清晰，各有侧重: 每个子策略都有明确的理论基础和逻辑，小市值（规模因子）、全天候（资产配置）、弱周期（高股息+防御）、ROA（深度价值）。
  3. 细节丰富，实战性强: 包含了处理A股特色（涨跌停、ST、除权除息）的大量代码，不是纸上谈兵的策略。
  4. 动态仓位管理: 特别是在弱周期策略中，根据市场估值动态调整股债比例，体现了主动风险管理思想。

潜在风险与改进点

  1. 小市值因子有效性: 小市值策略占比最高（40%），但其有效性在近年来A股市场受到挑战（龙头效应、注册制）。需密切关注其表现。
  2. 过往依赖: 策略中的一些参数和股票列表（如弱周期策略的初选池）基于历史经验，需要定期回顾和更新以适应市场变化。
  3. 回测未来函数: 在 calculatedividendratio 函数中，使用 enddate=self.context.currentdt 来查询分红数据。在回测中，currentdt 是回测当天的日期，但实际在当天开盘时，是无法知道当天是否会发布分红公告的。这引入了未来函数，会导致回测结果过于乐观。应改为 enddate=self.context.previous_date。
  4. 运行效率: 子策略较多，且有些选股范围较广（如全市场选股），在实盘运行时可能会遇到查询数据超时或运行缓慢的问题，可能需要优化代码结构。

总结

这是一个结构优秀、逻辑清晰、考虑周全的多策略量化组合。它不是一个追逐市场热点的策略，而更像一个基于经典投资理念（价值、规模、资产配置）构建的“压路机”，追求的是长期稳健的收益。其强大的风控和细节处理能力表明开发者具有丰富的实盘经验。

最大的亮点在于其组合思想：用进攻性的“小市值”作为矛，用防御性的“全天候”和“弱周期”作为盾，再用“简单ROA”作为卫星策略捕捉极端机会，构成了一个非常完整的投资体系。

如果您要使用此策略，建议重点关注小市值因子的表现并修复分红查询中的未来函数问题。

obsidian

📎 相关文章

📊 股票代码总索引

title股票代码总索引 created20260805 updated20260805 tags[股票代码, 索引, MOC] 📊 股票代码总索引 [!inf...

科技股下半年机会

date20260720 updated20260720 conversation_idf21364a807fd487da6923432916c6b30 tit...

社交数据 - 社交数据

社交数据 社交数据 共 36 张数据表 标签#数据字典 #13_社交数据 表列表 表名 中文名 字段数 说明 [[T_SOC]] 新三板代码表 7 [[T_SO...