本文目录导读:

综合实用脚本”和“变向突破次数对比”,这个问题通常出现在量化交易、自动化测试或游戏/AI行为模拟中。
为了给您最精准的解决方案,我将分两种最可能的情况进行解答。请根据您的实际场景选择(大概率是第一种量化交易场景):
量化交易 / 技术分析(最可能)
在股票或加密货币策略中,您想对比不同策略(或不同参数)下的“变向突破”频率(通常指唐奇安通道突破、布林带变轨或摆动指标)。
以下是一个基于 Python + Pandas 的综合脚本,用于计算并对比两种不同逻辑的“变向突破”次数:
import pandas as pd
import numpy as np
def calculate_breakout_counts(df, lookback=20, mode='close'):
"""
核心函数:计算变向突破次数(基于唐奇安通道)
参数:
df: 包含 'high', 'low', 'close' 的DataFrame
lookback: 回溯周期(即突破N日高点/低点)
mode: 'close' 表示收盘价突破,'intraday' 表示盘中价突破
返回:
dict: 包含上突破次数、下突破次数、总变向次数
"""
# 计算通道
df = df.copy()
df['upper'] = df['high'].rolling(window=lookback).max()
df['lower'] = df['low'].rolling(window=lookback).min()
# 判断突破方向(状态机:0=无突破,1=向上突破,-1=向下突破)
df['signal'] = 0
df.loc[df['close'] > df['upper'].shift(1), 'signal'] = 1
df.loc[df['close'] < df['lower'].shift(1), 'signal'] = -1
# 过滤无效值
df = df.dropna(subset=['upper', 'lower'])
# 计算“变向”次数(只有信号从非当前状态变为当前状态才计数)
df['regime'] = np.where(df['signal'] == 1, 1, np.where(df['signal'] == -1, -1, np.nan))
df['regime'] = df['regime'].ffill().fillna(0) # 前向填充,0为初值
# 检测变向:状态发生变化时计数
df['change'] = df['regime'].diff()
up_count = (df['change'] == 2).sum() # 从下变上(或空变上)
down_count = (df['change'] == -2).sum() # 从上变下(或空变下)
# 如果模式为intraday,则用high/low替代close做判断(示意)
if mode == 'intraday':
up_count = (df['high'] > df['upper'].shift(1)).sum()
down_count = (df['low'] < df['lower'].shift(1)).sum()
total = up_count + down_count
return {'up': up_count, 'down': down_count, 'total': total}
# ========== 实战对比示例 ==========
if __name__ == "__main__":
# 模拟生成数据(实际使用中请替换为真实数据源)
np.random.seed(42)
n = 500
price = np.cumsum(np.random.randn(n) * 1.5) + 100
df = pd.DataFrame({
'high': price + np.random.rand(n),
'low': price - np.random.rand(n),
'close': price
})
# 对比参数A(快线,10周期)与参数B(慢线,30周期)
result_fast = calculate_breakout_counts(df, lookback=10)
result_slow = calculate_breakout_counts(df, lookback=30)
# 输出对比报告
print("="*50)
print(f"策略A(快线10):向上{result_fast['up']}次, 向下{result_fast['down']}次, 总变向{result_fast['total']}次")
print(f"策略B(慢线30):向上{result_slow['up']}次, 向下{result_slow['down']}次, 总变向{result_slow['total']}次")
print("-"*50)
# 可视化对比(需matplotlib)
try:
import matplotlib.pyplot as plt
labels = ['向上突破', '向下突破', '总变向']
a = [result_fast['up'], result_fast['down'], result_fast['total']]
b = [result_slow['up'], result_slow['down'], result_slow['total']]
x = np.arange(len(labels))
width = 0.35
fig, ax = plt.subplots()
rects1 = ax.bar(x - width/2, a, width, label='快线(10)')
rects2 = ax.bar(x + width/2, b, width, label='慢线(30)')
ax.set_ylabel('次数')
ax.set_title('变向突破次数对比')
ax.set_xticks(x)
ax.set_xticklabels(labels)
ax.legend()
plt.show()
except ImportError:
print("(未安装matplotlib,跳过图表展示)")
游戏AI / 机器人 / 运动轨迹模拟
如果您指的是角色(或机器人)在移动中的变向频率(例如对比不同算法的路径平滑度),核心逻辑是检测方向向量的夹角突变。
def count_direction_changes(path_points, angle_threshold_deg=30):
"""
检测轨迹中的变向次数
path_points: 轨迹点列表 [(x1,y1), (x2,y2), ...]
angle_threshold_deg: 角度变化超过该值视为一次变向
返回: 变向次数
"""
if len(path_points) < 3:
return 0
def angle_between(v1, v2):
"""计算两个向量夹角(度)"""
dot = np.dot(v1, v2)
norm = np.linalg.norm(v1) * np.linalg.norm(v2)
if norm == 0:
return 0
cos_theta = np.clip(dot / norm, -1.0, 1.0)
return np.degrees(np.arccos(cos_theta))
changes = 0
prev_dir = None
for i in range(1, len(path_points)-1):
# 当前运动方向
v1 = np.array(path_points[i]) - np.array(path_points[i-1])
v2 = np.array(path_points[i+1]) - np.array(path_points[i])
if np.linalg.norm(v1) == 0 or np.linalg.norm(v2) == 0:
continue
angle = angle_between(v1, v2)
if angle > angle_threshold_deg:
changes += 1
return changes
# 对比两种路径
path_a = [...] # 算法A生成的路径
path_b = [...] # 算法B生成的路径
print(f"路径A变向次数: {count_direction_changes(path_a)}")
print(f"路径B变向次数: {count_direction_changes(path_b)}")
核心建议
- 明确“变向”定义:是价格/数据的“上穿下穿”,还是运动方向的“物理转角”?这直接决定了算法的核心逻辑。
- 数据清洗:如果是金融数据,务必使用前向填充避免未来函数;如果是运动数据,需先做平滑滤波(如卡尔曼滤波)去除噪声。
- 批次处理:如果数据量极大,建议将上述逻辑封装为类,并用
numba或numpy向量化加速。
如果您能告诉我:
- 数据长什么样?(列名、类型)
- “变向”的具体定义是什么?(突破N日高低点?角度超过X度?)
我可以直接为您定制一个更精确的脚本。