# 克隆自聚宽文章：https://www.joinquant.com/post/33636 # 标题：etf基金溢价-改进版-高收益低回撤-速度已最优 # 作者：发锅 # 核心改动就是调整了成交量检查标准变为过去10天最小值都要满足条件 # 本策略网址：https://www.joinquant.com/algorithm/live/index?backtestId=73d74ffd00f110ba66a454c11f10de93 # 导入函数库 from jqdata import * from jqlib.technical_analysis import * import numpy as np import pandas as pd import statsmodels.api as sm import datetime as dt import time # 初始化函数，设定基准等等 def initialize(context): # start_check_time = time.time() # 设定沪深300作为基准 set_benchmark('000300.XSHG') # 开启异步报单 set_option('async_order', True) # 开启动态复权模式(真实价格) set_option('use_real_price', True) # 是否未来函数 set_option("avoid_future_data", True) # 过滤掉order系列API产生的比error级别低的log # log.set_level('order', 'error') # 初始化全局变量 g.loss_limit = 0.9 # 单基金止损比例 g.drop_limit_days = 20 # 止损卖出后多少天不重新买入 g.control_days = 0 # 初始化控制全局止损之后暂停的天数 g.total_limit_days = 30 # 检查全局止损比例的天数范围 g.total_limit_rate = 0.15 # 全局止损比例 g.cool_days = 0 # 全局止损后多少天内不持仓，必须小于g.total_limit_days g.rate_list = [] g.check_loss_list = [] g.just_sell_list = [] g.total_value_list = [] g.hold_list = [] g.holiday = ['2010-02-12','2010-04-30','2010-09-30','2011-02-01','2011-04-29','2011-09-30','2012-01-20','2012-04-27','2012-09-28','2013-02-08', '2013-04-26','2013-09-30','2014-01-30','2014-04-30','2014-09-30','2015-02-17','2015-04-30','2015-09-30','2016-02-05','2016-04-29','2016-09-30', '2017-01-26','2017-04-28','2017-09-29','2018-02-14','2018-04-27','2018-09-28','2019-02-01','2019-04-30','2019-09-30','2020-01-23','2020-04-30', '2020-09-30','2021-02-10','2021-04-30','2021-09-30','2022-01-28','2022-04-29','2022-09-30','2023-01-20','2023-04-28','2023-09-28','2024-02-09', '2024-04-30','2024-09-30','2025-01-28','2025-04-30','2025-09-30','2026-02-15','2026-04-30','2026-09-30' ] set_order_cost(OrderCost(close_tax=0.000, open_commission=0.00025, close_commission=0.00025, min_commission=0), type='fund') set_slippage(PriceRelatedSlippage(0.002),type='fund') run_daily(before_market_open, '09:20', reference_security='000300.XSHG') run_daily(market_open, '09:30', reference_security='000300.XSHG') run_daily(check_loss_up, time='14:10', reference_security='000300.XSHG') # run_daily(print_position_info, time='15:10', reference_security='000300.XSHG') # end_check_time = time.time() # elapsed_time = end_check_time - start_check_time # print(f"initialize time: {elapsed_time}") def before_market_open(context): start_check_time = time.time() # 获取基金 fund_list = get_all_securities(['lof', 'etf'], context.previous_date).index.tolist() g.length1 = len(fund_list) # 过滤太新的基金 fund_list = filter_new_fund(context,fund_list) # 嘉实元和事件，所以在2019年5月之后不再买入 if context.current_dt.date() >= np.datetime64('2019-05-01') and ('505888.XSHG' in fund_list): fund_list.remove('505888.XSHG') print('remove 505888.XSHG') # 成交额过滤 df = history(count=10, unit='1d', field="money", security_list=fund_list).T # 整体花费时间少于0.16秒 df['min_money_10d'] = df.min(axis=1) # cur_total_value = context.portfolio.total_value # money_threshold = 2 * cur_total_value # print(f"cur_total_value: {cur_total_value}") money_threshold = 500000 # 固定值 df = df[df['min_money_10d'] > money_threshold] # 因为最大持仓是5种，那么资金就是5等分，那么只要是总金额的2倍，实际就是10倍 # 新的获取净值的方式，不是通过"get_extras"而是通过query和get_all_securities的方法 future_list = df.index.tolist() # 不需要去重，没有重复 temp_future_list = [item[:6] for item in future_list] # 和codes = list(set(temp_future_list))一致，也不存在重复 codes = list(set(temp_future_list)) # 获取当前日期的前10个交易日（包含当前日期） latest_trade_day = context.previous_date latest_trade_day = latest_trade_day.strftime('%Y-%m-%d') # 获取基金净值，目前除了159001以外，用for循环和in_的方式获得的数量一致 # 避免一次性获取数据量太大，进行分割查询 batch_size = 400 all_temp_dfs = [] # 按400个一组分割 temp_future_list 并逐组查询 for i in range(0, len(temp_future_list), batch_size): batch = temp_future_list[i:i + batch_size] temp_batch_df = finance.run_query( query(finance.FUND_NET_VALUE) .filter(finance.FUND_NET_VALUE.code.in_(batch)) .filter(finance.FUND_NET_VALUE.day==latest_trade_day) .order_by(finance.FUND_NET_VALUE.day.desc()) ) all_temp_dfs.append(temp_batch_df) # 合并所有批次的结果 temp_df = pd.concat(all_temp_dfs, ignore_index=True) # 获取所有基金的完整代码列表，并转为 DataFrame 便于匹配 all_funds = get_all_securities(['fund']) all_funds = all_funds.reset_index() # 重置索引，使 'index' 列成为 DataFrame 列 all_funds['code_prefix'] = all_funds['index'].str[:6] # 提取前6位数字 # 获取 temp_df 并提取 code 的前6位数字，方便匹配 temp_df['code_prefix'] = temp_df['code'].str[:6] # 将 temp_df 中的 code_prefix 匹配到 all_funds 中的 code_prefix，以获取完整的 code merged_df = pd.merge(temp_df, all_funds[['index', 'code_prefix']], left_on='code_prefix', right_on='code_prefix', how='left') # 检查是否存在未匹配的行，并打印出来 unmatched_df = merged_df[merged_df['index'].isna()] if not unmatched_df.empty: print("Unmatched entries:") print(unmatched_df[['code', 'code_prefix']]) # 使用完整的 code 作为索引，并重命名列为 'unit_net_value' df = merged_df.dropna(subset=['index']).set_index('index')[['net_value']].rename(columns={'net_value': 'unit_net_value'}) g.fund_list = df # 基金和净值的df # print(f"check g.fund_list structure: {type(g.fund_list)}, data: {g.fund_list}") log.info('开盘前记录净值...') end_check_time = time.time() elapsed_time = end_check_time - start_check_time print(f"before_market_open time: {elapsed_time}") def market_open(context): start_check_time = time.time() df = g.fund_list length2 = len(df) current = get_current_data() fund_list = df.index.tolist() ## 获得基金最新价 try: df['last_price'] = [current[c].last_price for c in fund_list] except Exception as e: print(f"error: {e}") # print("df: ") # print(df) ## 计算溢价 df['premium'] = (df.last_price / df.unit_net_value - 1) * 100 #最新价格小于净值的小于0 ## 根据溢价大小排序 if hasattr(df, 'sort'): # 如果有sort方法就用sort，没有用sort_values df = df.sort(['premium'], ascending = True) else: df = df.sort_values(['premium'], ascending = True) df = df[(df.premium < 0)] special_rate = len(df)/g.length1 # 最新价格低于净值的占比10%以内 g.rate_list.append(special_rate) g.rate_list = g.rate_list[-10:] # 最近10天的special_rate while len(g.rate_list) < 10: g.rate_list.append(g.rate_list[0]) print(f"g.rate_list - length: {len(g.rate_list)}, mean: {mean(g.rate_list)}") if g.cool_days == 0: if (len(g.rate_list) == 10) and (mean(g.rate_list) > 0.1): # 比例过低就不执行买入卖出的操作 target_fund_list = df[:30].index.tolist() target_fund_list = [stock for stock in target_fund_list if stock not in g.just_sell_list] target_fund_list = target_fund_list[:30] g.max_position = len(target_fund_list) # 卖出 for fund in context.portfolio.positions.keys(): # 卖出不在股票池或节假日前清仓 if fund not in target_fund_list or str(context.current_dt.date()) in g.holiday: order_target_value(fund, 0) # 买入, 节假日前不开仓 if str(context.current_dt.date()) not in g.holiday: # for fund in target_fund_list: # now_position = g.max_position - len(context.portfolio.positions) # if now_position == 0: # continue # if fund not in context.portfolio.positions.keys(): # position = context.portfolio.available_cash / now_position # order_target_value(fund, position) # 计算每个标的的10日平均成交额 money_df = history(count=10, unit='1d', field="money", security_list=target_fund_list).T money_df['avg_money'] = money_df.min(axis=1) # 过去10天成交额的最小值 # 计算单个持仓的标准金额 standard_position = context.portfolio.available_cash / g.max_position max_position_limit = standard_position * 3 # 单个标的最大持仓限制 print(f"max_position_limit: {max_position_limit}") for fund in target_fund_list: if fund not in context.portfolio.positions.keys(): # 计算该标的的目标买入金额 min_buy_amount = money_df.loc[fund, 'avg_money'] / 5 # 最小买入金额 target_amount = min(max_position_limit, max(min_buy_amount, standard_position)) print(f"{fund} min_buy_amount: {min_buy_amount} and target_amount: {target_amount}") # 确保不超过可用资金 if target_amount <= context.portfolio.available_cash: order_target_value(fund, target_amount) elif (len(g.rate_list) == 10) and (mean(g.rate_list) <= 0.1): if g.hold_list: clear_position(context) else: g.cool_days -= 1 # 更新持有的基金池 g.hold_list= [] for position in list(context.portfolio.positions.values()): fund = position.security g.hold_list.append(fund) end_check_time = time.time() elapsed_time = end_check_time - start_check_time print(f"market_open time: {elapsed_time}") ## 收盘后运行函数 def after_market_close(context): pass # 1-6 调整亏损比例过大的股票 def check_loss_up(context): start_check_time = time.time() if g.hold_list: check_loss_list = [] for stock in g.hold_list: position = context.portfolio.positions[stock] price = position.price avg_cost = position.avg_cost # print('check %s, price: %2f, avg_cost: %2f' % (stock, price, avg_cost)) if price < g.loss_limit * avg_cost: log.info("[%s]损失比例过高，卖出" % stock) close_position(position) check_loss_list.append(stock) if check_loss_list: g.check_loss_list.append(check_loss_list) else: g.check_loss_list.append(['nothing']) if len(g.check_loss_list) > g.drop_limit_days: g.check_loss_list = g.check_loss_list[-g.drop_limit_days:] temp_set = set() for check_loss_list in g.check_loss_list: temp_set = temp_set.union(set(check_loss_list)) # 不要购买的股票列表，过往20天因为止损而卖出的股票 g.just_sell_list = list(temp_set) check_total_value(context) end_check_time = time.time() elapsed_time = end_check_time - start_check_time print(f"check_loss_up time: {elapsed_time}") # 1-7 检查整体资金比例 def check_total_value(context): start_check_time = time.time() total_money_today = context.portfolio.total_value g.total_value_list.append(total_money_today) print('检查整体资金比例g.total_value_list: ', len(g.total_value_list)) print(g.total_value_list) if len(g.total_value_list) >= g.total_limit_days: g.total_value_list = g.total_value_list[-g.total_limit_days:] # 只考虑最近20天的跌幅来判断是否清仓 biggest_pullback = (total_money_today - max(g.total_value_list))/max(g.total_value_list) print('检查近 %d 的最大损失为 %2f' % (g.total_limit_days, biggest_pullback * 100)) if biggest_pullback < - g.total_limit_rate: # 当跌幅超过最大限制，则清空仓位 clear_position(context) if g.control_days == 0: # 设定空仓天数 print('清仓后，未修正的g.control_days为: ', g.control_days) g.control_days = g.cool_days print('清仓后，修正g.control_days为: ', g.control_days) print('持仓情况为: ', g.hold_list) print('判断标准为: ', (not g.hold_list)) if not g.hold_list: # 如果卖光了，那么调整检查全盘资金的数据量，保留10天的数据，因为检查是最近20天，暂停10天 g.total_value_list = g.total_value_list[-(g.total_limit_days-g.cool_days):] end_check_time = time.time() elapsed_time = end_check_time - start_check_time print(f"check_total_value time: {elapsed_time}") #3-1 交易模块-自定义下单 def order_target_value_(security, value): if value == 0: log.debug("Selling out %s" % (security)) else: log.debug("Order %s to value %f" % (security, value)) return order_target_value(security, value) #3-3 交易模块-平仓 def close_position(position): security = position.security order = order_target_value_(security, 0) # 可能会因停牌失败 if order != None: if order.status == OrderStatus.held and order.filled == order.amount: return True return False #3-5 交易模块 - 清仓 def clear_position(context): if context.portfolio.positions: g.cool_days = 5 # 清仓后5天不进行买入操作 log.info("==> 清仓，卖出所有股票") for stock in context.portfolio.positions.keys(): position = context.portfolio.positions[stock] close_position(position) #2-7 过滤次新股 def filter_new_fund(context,stock_list): start_check_time = time.time() yesterday = context.previous_date end_check_time = time.time() elapsed_time = end_check_time - start_check_time print(f"filter_new_fund time: {elapsed_time}") return [stock for stock in stock_list if not yesterday - get_security_info(stock).start_date < datetime.timedelta(days=5)] # 清理list里nan的模块 def clean_List_nan(List): Myarray=np.array(List) x = float('nan') for elem in Myarray: if math.isnan(x): x = 0.0 return Myarray #4-1 打印每日持仓信息 def print_position_info(context): #打印当天成交记录 trades = get_trades() for _trade in trades.values(): print('成交记录：'+str(_trade)) #打印账户信息 for position in list(context.portfolio.positions.values()): securities=position.security cost=position.avg_cost price=position.price ret=100*(price/cost-1) value=position.value amount=position.total_amount print('代码:{}'.format(securities)) print('成本价:{}'.format(format(cost,'.2f'))) print('现价:{}'.format(price)) print('收益率:{}%'.format(format(ret,'.2f'))) print('持仓(股):{}'.format(amount)) print('市值:{}'.format(format(value,'.2f'))) print('———————————————————————————————————') # 5-1 临时检查的一些函数，查获取基金净值的方法有没有漏洞 def check_fund_net_value(result_df, result_temp_df): # 1. 比较每个 code 的记录数差异 # 按 code 统计每个查询结果的记录数 result_counts_1 = result_df.groupby('code').size() result_counts_2 = result_temp_df.groupby('code').size() print(f"result_counts_1: {result_counts_1.shape}, result_counts_2: {result_counts_2.shape}") # 将 result_counts_1 和 result_counts_2 转换为 DataFrame 并进行 join 对齐 result_counts_df = pd.DataFrame({ 'count_1': result_counts_1, 'count_2': result_counts_2 }).fillna(0) # 使用 0 填充缺失值 # 找到记录数不一致的 code mismatched_codes = result_counts_df[result_counts_df['count_1'] != result_counts_df['count_2']].index.tolist() # print(f"Records mismatch for codes: {mismatched_codes}") # 找出在两种方法中缺失的 code missing_in_result_temp_df = list(set(result_counts_1.index) - set(result_counts_2.index)) missing_in_result_df = list(set(result_counts_2.index) - set(result_counts_1.index)) print(f"Codes missing in result_temp_df: {missing_in_result_temp_df}") print(f"Codes missing in result_df: {missing_in_result_df}") # 创建一个包含 'code' 和 'day' 的 DataFrame，用于标记存在情况 all_codes_days = pd.DataFrame({ 'code': pd.concat([result_df['code'], result_temp_df['code']]), 'day': pd.concat([result_df['day'], result_temp_df['day']]) }).drop_duplicates() # 使用 merge 操作标记 result_df 和 result_temp_df 中的数据存在情况 all_codes_days = all_codes_days.merge( result_temp_df[['code', 'day']], on=['code', 'day'], how='left', indicator='in_result_temp_df' ).merge( result_df[['code', 'day']], on=['code', 'day'], how='left', indicator='in_result_df' ) # 添加标志列 all_codes_days['in_result_temp_df'] = (all_codes_days['in_result_temp_df'] == 'both').astype(int) all_codes_days['in_result_df'] = (all_codes_days['in_result_df'] == 'both').astype(int) # 去掉第3和第4列都为1的数据 filtered_df = all_codes_days[~((all_codes_days['in_result_temp_df'] == 1) & (all_codes_days['in_result_df'] == 1))] # 打印结果 # print(filtered_df) for fund in missing_in_result_temp_df: print(f"check {fund}") test_df1 = finance.run_query( query(finance.FUND_NET_VALUE) .filter(finance.FUND_NET_VALUE.code == fund) .order_by(finance.FUND_NET_VALUE.day.desc()) ) temp_list2 = [fund] test_df2 = finance.run_query(query(finance.FUND_NET_VALUE).filter(finance.FUND_NET_VALUE.code.in_(temp_list2)).order_by(finance.FUND_NET_VALUE.day.desc())) test_df3 = test_df2.groupby('code').head(10).reset_index(drop=True) print(f"test_df1: {test_df1.shape}") print(f"test_df3: {test_df3.shape}") print(f"test_df2: {test_df2.shape}")