174 lines
4.5 KiB
Python
174 lines
4.5 KiB
Python
"""
|
|
基础功能测试
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
|
|
# 添加项目根目录到路径
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from data.data_processor import DataProcessor
|
|
from core.kline import KLine
|
|
from core.fractal import Fractal
|
|
from core.chan_analyzer import ChanAnalyzer
|
|
|
|
|
|
def create_test_data():
|
|
"""创建测试用的K线数据"""
|
|
dates = pd.date_range(start='2024-01-01', periods=100, freq='H')
|
|
|
|
# 生成模拟价格数据
|
|
np.random.seed(42)
|
|
base_price = 50000
|
|
prices = [base_price]
|
|
|
|
for i in range(99):
|
|
change = np.random.normal(0, 100) # 价格变化
|
|
new_price = max(prices[-1] + change, 1000) # 确保价格为正
|
|
prices.append(new_price)
|
|
|
|
# 生成OHLCV数据
|
|
data = []
|
|
for i, date in enumerate(dates):
|
|
if i == 0:
|
|
open_price = prices[i]
|
|
else:
|
|
open_price = data[-1]['close']
|
|
|
|
close_price = prices[i]
|
|
high_price = max(open_price, close_price) + np.random.uniform(0, 50)
|
|
low_price = min(open_price, close_price) - np.random.uniform(0, 50)
|
|
volume = np.random.uniform(1000, 10000)
|
|
|
|
data.append({
|
|
'open': open_price,
|
|
'high': high_price,
|
|
'low': low_price,
|
|
'close': close_price,
|
|
'volume': volume
|
|
})
|
|
|
|
df = pd.DataFrame(data, index=dates)
|
|
return df
|
|
|
|
|
|
def test_data_processor():
|
|
"""测试数据处理器"""
|
|
print("测试数据处理器...")
|
|
|
|
# 创建测试数据
|
|
test_data = create_test_data()
|
|
|
|
# 验证数据
|
|
processor = DataProcessor()
|
|
is_valid = processor.validate_klines(test_data)
|
|
print(f"数据验证结果: {is_valid}")
|
|
|
|
# 清理数据
|
|
cleaned_data = processor.clean_klines(test_data)
|
|
print(f"清理后数据量: {len(cleaned_data)}")
|
|
|
|
# 添加技术指标
|
|
data_with_indicators = processor.add_technical_indicators(cleaned_data)
|
|
print(f"技术指标列: {list(data_with_indicators.columns)}")
|
|
|
|
return cleaned_data
|
|
|
|
|
|
def test_kline_processor():
|
|
"""测试K线处理器"""
|
|
print("\n测试K线处理器...")
|
|
|
|
test_data = create_test_data()
|
|
|
|
# K线包含关系处理
|
|
kline_processor = KLine(test_data)
|
|
processed_data = kline_processor.get_processed_data()
|
|
|
|
print(f"原始K线数量: {len(test_data)}")
|
|
print(f"处理后K线数量: {len(processed_data)}")
|
|
|
|
# 可视化信息
|
|
viz_info = kline_processor.visualize_containment()
|
|
print(f"合并统计: {viz_info}")
|
|
|
|
return processed_data
|
|
|
|
|
|
def test_fractal_detector():
|
|
"""测试分型识别器"""
|
|
print("\n测试分型识别器...")
|
|
|
|
processed_data = test_kline_processor()
|
|
|
|
# 分型识别
|
|
fractal_detector = Fractal(processed_data, min_strength=1)
|
|
fractals = fractal_detector.detect_fractals()
|
|
|
|
print(f"检测到分型数量: {len(fractals)}")
|
|
|
|
# 分型统计
|
|
stats = fractal_detector.get_fractal_statistics()
|
|
print(f"分型统计: {stats}")
|
|
|
|
return fractals
|
|
|
|
|
|
def test_chan_analyzer():
|
|
"""测试综合分析器"""
|
|
print("\n测试综合分析器...")
|
|
|
|
test_data = create_test_data()
|
|
|
|
# 完整分析
|
|
analyzer = ChanAnalyzer(test_data)
|
|
summary = analyzer.run_full_analysis(fractal_strength=1)
|
|
|
|
print("分析结果摘要:")
|
|
for key, value in summary.items():
|
|
print(f" {key}: {value}")
|
|
|
|
# 获取最新信号
|
|
latest_signals = analyzer.get_latest_signals(24)
|
|
print(f"\n最新信号数量: {len(latest_signals)}")
|
|
|
|
# 市场结构
|
|
market_structure = analyzer.get_current_market_structure()
|
|
print(f"当前市场结构: {market_structure}")
|
|
|
|
return analyzer
|
|
|
|
|
|
def run_all_tests():
|
|
"""运行所有测试"""
|
|
print("=" * 50)
|
|
print("缠论分析系统基础功能测试")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
# 测试各个模块
|
|
test_data_processor()
|
|
test_kline_processor()
|
|
test_fractal_detector()
|
|
analyzer = test_chan_analyzer()
|
|
|
|
print("\n" + "=" * 50)
|
|
print("所有测试完成！")
|
|
print("=" * 50)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\n测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = run_all_tests()
|
|
sys.exit(0 if success else 1) |