问题描述
请问有没有大神能编一个在一定区域内,可以找出与当下最高k线相比(且需判断当下这根k线已创新高),排在第二高那根k线的代码?
之前见过大神改编的MACD金叉死叉代码,是用MACD柱子穿越0轴作为index标记位置的,但这样会使得标记点随着参数改变而改变
不知道有没有那种可以不依赖参数就能得出某段时间内第二极值的代码?
非常感谢
解决方案
在 JoinQuant 平台上,寻找一定区域内的次高点或次低点(第二极值)并不需要依赖 MACD 等技术指标的参数。我们可以纯粹基于 K 线的价格数据(如最高价 high 或最低价 low),利用 Python 的 Pandas 库进行排序和筛选,从而精准地找到第二极值。
以下是一个完整的代码示例,封装成了一个函数。该函数会获取指定标的在过去 N 根 K 线内的数据,判断当前 K 线是否创出新高,并找出这段时间内的次高点(第二高点)及其对应的时间。
Python 实现代码
import pandas as pd
from jqdata import *
def find_second_extremes(security, n, unit='1d'):
"""
寻找指定区域内的最高点、次高点,并判断当前K线是否创新高
:param security: 标的代码,如 '000001.XSHE'
:param n: 观察的 K 线数量(包含当前 K 线)
:param unit: K 线周期,如 '1d', '60m', '5m'
:return: 包含极值信息的字典
"""
# 使用 get_bars 获取包含当前时刻在内的过去 n 根 K 线数据
# include_now=True 确保包含了当下的这根 K 线
bars = get_bars(security, count=n, unit=unit, fields=['date', 'high', 'low'], include_now=True, df=True)
if len(bars) < 2:
print("K线数量不足,无法比较")
return None
# ---------------- 寻找最高点与次高点 ----------------
# 按最高价降序排列
sorted_highs = bars.sort_values(by='high', ascending=False)
# 最高点(排在第一)
highest_bar = sorted_highs.iloc[0]
# 次高点(排在第二)
second_highest_bar = sorted_highs.iloc[1]
# ---------------- 寻找最低点与次低点 ----------------
# 按最低价升序排列
sorted_lows = bars.sort_values(by='low', ascending=True)
# 最低点(排在第一)
lowest_bar = sorted_lows.iloc[0]
# 次低点(排在第二)
second_lowest_bar = sorted_lows.iloc[1]
# ---------------- 判断当下是否创新高/新低 ----------------
# 获取当下的这根 K 线(即列表中的最后一根)
current_bar = bars.iloc[-1]
# 如果最高点的时间等于当前 K 线的时间,说明当下这根 K 线创了新高
is_new_high = (current_bar['date'] == highest_bar['date'])
# 同理判断是否创新低
is_new_low = (current_bar['date'] == lowest_bar['date'])
return {
'current_date': current_bar['date'],
'is_new_high': is_new_high,
'highest_val': highest_bar['high'],
'highest_date': highest_bar['date'],
'second_highest_val': second_highest_bar['high'],
'second_highest_date': second_highest_bar['date'],
'is_new_low': is_new_low,
'lowest_val': lowest_bar['low'],
'lowest_date': lowest_bar['date'],
'second_lowest_val': second_lowest_bar['low'],
'second_lowest_date': second_lowest_bar['date']
}
如何在策略中使用?
你可以在 handle_data 或定时运行的函数中调用这个方法。例如,你想在每天开盘时检查过去 20 天内的情况:
def initialize(context):
g.security = '000001.XSHE'
run_daily(market_open, time='09:30')
def market_open(context):
# 查找过去 20 根日线的极值情况
result = find_second_extremes(g.security, n=20, unit='1d')
if result:
if result['is_new_high']:
log.info(f"【创新高】当前K线创出近20天新高!最高价: {result['highest_val']}")
log.info(f"排在第二高的K线出现在 {result['second_highest_date']},价格为: {result['second_highest_val']}")
# 在这里编写你的交易逻辑,例如:
# 如果当前价格突破了次高点一定比例,则买入等...
else:
log.info("当前K线未创新高。")
代码原理解析
get_bars函数:我们使用get_bars并设置include_now=True,这样可以获取到包含当前正在运行的这根 K 线在内的历史数据。这比attribute_history更适合用于判断“当下”的状态。- Pandas 排序:利用
sort_values对high列进行降序排列,排在第 0 位的自然是最高点,排在第 1 位的自然是次高点。这种方法完全基于价格本身,不依赖任何技术指标参数。 - 判断创新高:通过比对“最高点所在的时间”和“当前 K 线的时间”是否一致,就能准确判断出当前这根 K 线是否是这段区域内的最高点(即是否创新高)。