brunoandClaude
b28e9aa308
- 移动调试脚本到 debug_scripts/ 目录 - 清理 Python 缓存文件和编译文件 - 添加综合数据调度器和相关文档 - 整理项目结构，提高代码可维护性 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>
90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import sys
|
|
import os
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
|
|
# 测试策略函数
|
|
def test_triple_ma_strategy(data: pd.DataFrame, ma5=5, ma10=10, ma20=20, volume_threshold=1.2, **kwargs):
|
|
"""三均线策略测试"""
|
|
print(f"测试参数: ma5={ma5}, ma10={ma10}, ma20={ma20}")
|
|
print(f"数据形状: {data.shape}")
|
|
print(f"数据列: {data.columns.tolist()}")
|
|
|
|
signals = pd.DataFrame(index=data.index, columns=['signal', 'signal_strength'])
|
|
signals['signal'] = 0
|
|
signals['signal_strength'] = 0.0
|
|
|
|
# 确保参数是整数
|
|
ma5, ma10, ma20 = int(ma5), int(ma10), int(ma20)
|
|
print(f"转换后参数: ma5={ma5}, ma10={ma10}, ma20={ma20}")
|
|
|
|
# 计算均线
|
|
ma_5 = data['close'].rolling(window=ma5).mean()
|
|
ma_10 = data['close'].rolling(window=ma10).mean()
|
|
ma_20 = data['close'].rolling(window=ma20).mean()
|
|
|
|
print(f"MA5计算完成，长度: {len(ma_5)}")
|
|
print(f"MA10计算完成，长度: {len(ma_10)}")
|
|
print(f"MA20计算完成，长度: {len(ma_20)}")
|
|
|
|
# 成交量均线
|
|
vol_ma = data['volume'].rolling(window=20).mean()
|
|
|
|
signal_count = 0
|
|
for i in range(max(ma5, ma10, ma20), len(data)):
|
|
# 多头排列且放量
|
|
if (ma_5.iloc[i] > ma_10.iloc[i] > ma_20.iloc[i] and
|
|
data['volume'].iloc[i] > vol_ma.iloc[i] * volume_threshold):
|
|
signals.iloc[i, signals.columns.get_loc('signal')] = 1
|
|
signals.iloc[i, signals.columns.get_loc('signal_strength')] = 0.8
|
|
signal_count += 1
|
|
# 空头排列
|
|
elif ma_5.iloc[i] < ma_10.iloc[i] < ma_20.iloc[i]:
|
|
signals.iloc[i, signals.columns.get_loc('signal')] = -1
|
|
signals.iloc[i, signals.columns.get_loc('signal_strength')] = 0.8
|
|
signal_count += 1
|
|
|
|
print(f"生成信号数量: {signal_count}")
|
|
print(f"信号统计:")
|
|
print(signals['signal'].value_counts())
|
|
|
|
return signals
|
|
|
|
if __name__ == "__main__":
|
|
# 创建模拟数据
|
|
dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
|
|
np.random.seed(42)
|
|
|
|
# 生成模拟股价数据
|
|
n_days = len(dates)
|
|
base_price = 10.0
|
|
returns = np.random.normal(0.001, 0.02, n_days) # 0.1%日均收益，2%波动率
|
|
prices = [base_price]
|
|
|
|
for ret in returns[1:]:
|
|
prices.append(prices[-1] * (1 + ret))
|
|
|
|
data = pd.DataFrame({
|
|
'date': dates,
|
|
'open': prices,
|
|
'high': [p * (1 + abs(np.random.normal(0, 0.01))) for p in prices],
|
|
'low': [p * (1 - abs(np.random.normal(0, 0.01))) for p in prices],
|
|
'close': prices,
|
|
'volume': np.random.uniform(1000000, 5000000, n_days),
|
|
'pct_chg': [0] + [((prices[i] - prices[i-1]) / prices[i-1] * 100) for i in range(1, len(prices))]
|
|
})
|
|
data.set_index('date', inplace=True)
|
|
|
|
print("开始测试策略函数...")
|
|
print("=" * 50)
|
|
|
|
# 测试策略
|
|
result = test_triple_ma_strategy(data, ma5=5, ma10=10, ma20=20, volume_threshold=1.2)
|
|
|
|
print("=" * 50)
|
|
print("测试完成") |