图片合法配资平台官网
今天分享的源码是一个基于量化掘金平台的黄金坑选股策略。策略的找出黄金坑的部分的代码是从知乎的一篇博文中看到的，但是不是基于掘金量化平台，而且是日线级别的选股。我在此基础上修改为在掘金量化平台运行的，而且可以从不同级别来选股的，包括日线、周线、月线、季线和年线级别。当前代码最多只选出9只。有兴趣的朋友可以修改代码来选出更多票。不多说了，下面上干货。这个选股策略比较完整，没有任何私域代码。
# coding=utf-8from gm.api import *from datetime import datetimefrom datetime import timedeltaimport talibimport numpy as npfrom collections import dequeimport pandas as pdimport matplotlib.pyplot as plt# 本策略基于掘金量化交易平台 网址：www.myquant.cn# 常用参量设置DATE_STR = '%Y-%m-%d'TIME_STR = '%Y-%m-%d %H:%M:%S'SHORT_PERIOD = 5LONG_PERIOD = 20# D W M Q 四种，分别是日，周，月，季FREQUENCY = 'W'def init(context): # 全局变量设置 context.dict_stock_price = dict() context.symbol_pool = get_instruments_all_stocks(context) # 日期设定，避免出现未来函数，将起始日往前取一日 context.end_date =datetime.strptime(context.backtest_start_time, TIME_STR) context.start_date = datetime.strftime(context.end_date - timedelta(days=1000), TIME_STR) stockList = [] print(find_parabola(context))def get_instruments_all_stocks(context): ''' 查询沪深A股当天有交易的股票代码信息 ''' try: # 获取上一个交易日的日期 last_day = get_previous_trading_date(exchange='SHSE', date=context.now) # 获取正常交易的股票信息 trading_stocks, all_stocks_str = get_normal_stocks(last_day) return trading_stocks except Exception as e: # 对可能的异常进行捕获和处理，例如记录日志或返回一个错误信息 # 实际使用中应该根据具体情况进行定制 print(f'Error occurred: {e}') # 根据函数的要求，遇到异常时是否返回空列表或抛出异常是需要确定的 return [] # 或者可以选择抛出异常而不是返回空列表def get_normal_stocks(date,new_days=365): ''' 获取目标日期date的A股代码（剔除停牌股、ST股、次新股（365天）。剔除9238开头的股票） :param date：目标日期 :param new_days:新股上市天数，默认为365天 ''' if isinstance(date,str) and len(date)==10: date = datetime.strptime(date,'%Y-%m-%d').date() elif isinstance(date,str) and len(date)>10: date = datetime.strptime(date,'%Y-%m-%d %H:%M:%S').date() # 先剔除退市股、次新股和B股 科创板 df_code = get_instrumentinfos(sec_types=SEC_TYPE_STOCK, fields='symbol, listed_date, delisted_date', df=True) all_stocks = [code for code in df_code[(df_code['listed_date'].dt.date<=date-timedelta(days=new_days)) &(df_code['delisted_date'].dt.date>date)].symbol.to_list() if code[:6]!='SHSE.9' and code[:6]!='SZSE.2' and code[:6]!='SZSE.3' and code[:6]!='SZSE.8' and code[:7]!='SHSE.68'] # 再剔除当前的停牌股和ST股 history_ins = get_history_instruments(symbols=all_stocks, start_date=date, end_date=date, fields='symbol,sec_level, is_suspended', df=True) if len(history_ins)<=0 : print('未查到数据！') return [],[] all_stocks = list(history_ins[(history_ins['sec_level']==1) & (history_ins['is_suspended']==0)]['symbol']) all_stocks_str = ','.join(all_stocks) print('{}获取沪深所有在交易的股票代码完毕！获取数量：{}'.format(date, len(all_stocks))) return all_stocks,all_stocks_strdef find_adv_maline(context, slp=0.03): ''' 查找月线回踩的股票 :param end: 截止日期 :param slp: 斜率 默认0.04 :param securities: 股票池，只包含股票代码 :return: ''' resultList = [] import arrow _end = arrow.get(context.end_date) for code in context.stock_pool: if code not in context.stock_pool: print('Warning: cannot obtain price of stock {} at date {}'.format( code, context.now)) bar = compose_bar(code, context.start_date, context.backtest_end_time, FREQUENCY)[-15:] all_up = np.all(bar['close'][-3:] > bar['open'][-3:]) ma = moving_average(bar['close'], 5) try: if slope(ma, 0.3) >= slp and all_up: resultList.append(code) print(code) except Exception: pass context.resultList = resultListif __name__ == '__main__': run(strategy_id='b83232b7-fce2-11eb-aa99-1ac43a38feea', filename='main4.py', mode=MODE_BACKTEST, backtest_adjust=ADJUST_PREV, token='2a73fd387f79b2c096806a91a6b13f915f981007', backtest_start_time='2025-04-10 09:00:00', backtest_end_time='2025-04-11 15:30:00')def moving_average(ts, win): ''' 计算平均线 :param ts: 时间序列 :param win: 日 :return: ''' return np.convolve(ts, np.ones(win) / win, 'valid')def rmse(y, y_hat): ''' 返回预测序列相对于真值序列的标准差。 Args: y: y_hat: Returns: ''' return np.sqrt(np.mean(np.square(y - y_hat)))def slope(ts, err): ''' 返回直线斜率。如果拟合误差大于err，则抛出异常 ''' # 对ts进行归一化，以便斜率可以在不同的时间序列之间进行比较 assert ts[0] != 0 ts = ts / ts[0] x = np.arange(len(ts)) z = np.polyfit(x, ts, deg=1) # print(z) p = np.poly1d(z) ts_hat = np.array([p(xi) for xi in x]) error = rmse(ts, ts_hat) / np.sqrt(np.mean(np.square(ts))) if error >= err: raise ValueError('can not fit into line') return z[0]def find_runs(x): '''Find runs of consecutive items in an array.''' # ensure array x = np.asanyarray(x) if x.ndim != 1: raise ValueError('only 1D array supported') n = x.shape[0] # handle empty array if n == 0: return np.array([]), np.array([]), np.array([]) else: # find run starts loc_run_start = np.empty(n, dtype=bool) loc_run_start[0] = True np.not_equal(x[:-1], x[1:], out=loc_run_start[1:]) run_starts = np.nonzero(loc_run_start)[0] # find run values run_values = x[loc_run_start] # find run lengths run_lengths = np.diff(np.append(run_starts, n)) return run_values, run_starts, run_lengthsdef is_long_parallel(ts, n): ''' :param ts: 收盘价数组 :params n: 多头排列刚形成n天 ''' ma5 = moving_average(ts, 5) ma10 = moving_average(ts, 10) ma20 = moving_average(ts, 20) # 注意各均线序列的长度并不一样 ma5 = ma5[-10:] ma10 = ma10[-10:] ma20 = ma20[-10:] if len(ma5) == len(ma10) & len(ma10) == len(ma20): signal = (ma5 > ma10) & (ma10 > ma20) run_values, run_starts, run_lengths = find_runs(signal) # 最后一段必须为多头排列，且长度为要求的n return run_values[-1] == True and run_lengths[-1] == n else: return Falsedef find_parabola(context, unit='1d', a=0, e=3e-3, fit_win=7): ''' 找出黄金坑 :param unit: :param a: :param e: :param fit_win: :return: ''' x = np.arange(7) # 设置与显示相关的参数 plt.figure(figsize=(9, 9)) plt.subplots_adjust(hspace=0.3) results = [] count = 0 for stock in context.symbol_pool: bars = compose_bar(stock, context.start_date, context.end_date, FREQUENCY)[-(fit_win + 5 - 1):] if len(bars) != fit_win + 5 - 1: continue ma = moving_average(bars['close'], 5) ts = ma / ma[0] coef = np.polyfit(x, ts, deg=2) p = np.poly1d(coef) ts_hat = np.array([p(xi) for xi in x]) err = rmse(ts, ts_hat) / np.sqrt(np.mean(np.square(ts))) if err > e: continue _a, _b, _c = coef vert_x = -_b / (2 * _a) t1 = _a > a and (fit_win - 2) > vert_x > 2 # 如果是用于实际选股，你还需要加上其它条件，比如当前收阳，或者量能增加等 if not t1: continue vol_up = bars['volume'].to_numpy(dtype='float')[-4:-1] > bars['volume'].to_numpy(dtype='float')[-5:-2] t2 = np.count_nonzero(vol_up) >= 3 if not t2: continue ups = bars['close'].to_numpy(dtype='float')[-3:] > bars['close'].to_numpy(dtype='float')[-4:-1] t3 = np.count_nonzero(ups) == 3 if not t3: continue # 作为示例，我们只输出9张图 if count >= 9: break if t1: count += 1 results.append(stock) ax = plt.subplot(int(f'33{count}')) # 显示股票代码，为避免推广嫌疑，这里注释掉。读者在运行时可以自己打开。 ax.set_title(stock) ax.set_title(f'a:{_a:.3f} b:{_b:.3f}, vx: {vert_x:.1f}') plt.plot(x, ts) plt.plot(x, ts_hat, '--') plt.show() return resultsdef compose_bar(symbol, start_time, end_time, frequency='W'): ''' symbol:标的， start_time:开始时间， end_time:结束时间， frequency:频率 ''' # 合成周线 if frequency == 'W': # 判断开始日期是否为周一（默认周一 = 0，周二 = 1，所以对得出的week进行调整，加一） start_time = datetime.strptime(start_time, TIME_STR) week_start = start_time.weekday() + 1 # 如果不是周一，则将开始时间调整到该周周一 if week_start != 1: # print('输入的开始日期为周{}，调整到该周周一'.format(week_start)) start_time = start_time - timedelta(days=week_start - 1) # 合成月线 if frequency == 'M': # 判断开始日期是否为月初 day_start_1th = int(start_time[8:10]) # 如果不是1号，则转为当月1号 if day_start_1th != 1: start_time = datetime.strptime(start_time, TIME_STR) start_time = start_time - timedelta(days=day_start_1th - 1) # 合成季线 if frequency == 'Q': # 直接开始日期设置为季度初 month = int(start_time[5:7]) # 看开始时间月份 year = int(start_time[0:4]) # 看开始时间年份 if month in range(1, 3): start_time = datetime.date(year, 1, 1) if month in range(4, 6): start_time = datetime.date(year, 4, 1) if month in range(7, 9): start_time = datetime.date(year, 7, 1) if month in range(10, 12): start_time = datetime.date(year, 9, 1) # 合成年线 if frequency == 'A': # 将开始时间调整为年初 year_start = int(start_time[0:4]) # 看开始时间年份 start_time = datetime.date(year_start, 1, 1) # 订阅历史数据（注意要复权到当前日期才能和新浪财经的数据对得上） data = history(symbol=symbol, frequency='1d', start_time=start_time, end_time=end_time, fields='eob,open,close,high,low,amount,volume', df=True, skip_suspended=True, fill_missing=None, adjust=ADJUST_PREV, adjust_end_time=end_time) if len(data) > 0: # 修改日期格式并变成索引 data['eob'] = data['eob'].apply(lambda x: datetime.strptime(str(x).split(' ')[0], '%Y-%m-%d')) data.set_index(data['eob'], inplace=True) data.drop(columns=['eob'], inplace=True) data_index = data.resample(frequency, label='right').last().index # 结果返回成dataframe格式 data_k = pd.DataFrame({'open': data.resample(frequency, label='right').first()['open'], 'close': data.resample(frequency, label='right').last()['close'], 'high': data.resample(frequency, label='right').max()['high'], 'low': data.resample(frequency, label='right').min()['low'], 'amount': data.resample(frequency, label='right').sum()['amount'], 'volume': data.resample(frequency, label='right').sum()['volume'], 'frequency': frequency }) data_k.set_index(data_index, inplace=True) # 如果某周放假没有数据，则删除 data_k.dropna(inplace=True) return data_k else: return pd.DataFrame( columns=['open', 'close', 'high', 'low', 'amount', 'volume', 'frequency']) # 创建一个空的dataframe这是一个基于掘金量化交易平台的股票选股策略代码。让我分几个主要部分来解释：
初始化和设置
# 常用参量设置DATE_STR = '%Y-%m-%d'TIME_STR = '%Y-%m-%d %H:%M:%S'SHORT_PERIOD = 5LONG_PERIOD = 20FREQUENCY = 'W' # 周线数据
这部分设置了常用的时间格式和周期参数合法配资平台官网，使用周线数据进行分析。
主要功能函数:
init(context): 策略初始化函数，设置全局变量和股票池get_instruments_all_stocks(context): 获取沪深A股当天有交易的股票代码get_normal_stocks(date, new_days=365): 获取正常交易的股票，剔除停牌股、ST股、次新股等find_adv_maline(context, slp=0.03): 查找月线回踩的股票find_parabola(context, unit='1d', a=0, e=3e-3, fit_win=7): 找出月线黄金坑形态的股票compose_bar(symbol, start_time, end_time, frequency='W'): 合成K线数据（支持日、周、月、季、年线）技术指标计算函数:
moving_average(ts, win): 计算移动平均线rmse(y, y_hat): 计算均方根误差slope(ts, err): 计算序列的斜率find_runs(x): 查找连续序列is_long_parallel(ts, n): 判断是否形成多头排列策略核心逻辑:
使用抛物线拟合来寻找黄金坑形态通过均线系统判断趋势结合成交量分析使用斜率分析判断趋势强度主要选股条件:
抛物线拟合误差小于阈值成交量连续放大价格连续上涨均线多头排列数据获取和处理:
使用掘金量化平台的历史数据接口支持不同周期的K线数据合成自动处理停牌、复权等问题这个策略的主要目标是寻找具有以下特征的股票：
形成黄金坑形态（抛物线底部）成交量开始放大价格开始上涨均线系统呈现多头排列策略使用了多种技术分析方法，包括：
均线系统分析成交量分析形态分析（抛物线拟合）趋势分析（斜率计算）需要注意的是合法配资平台官网，这个策略是在掘金量化平台上运行的，需要相应的API token和权限才能执行。策略的回测时间范围是从2025年4月10日到2025年4月11日。
本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请点击举报。倍顺网配资提示：文章来自网络，不代表本站观点。
相关文章
沪深京指数
热点资讯
