179 lines
6.5 KiB
Python
179 lines
6.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试价格回踩阴线最高点提醒功能
|
||
"""
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
import pandas as pd
|
||
from datetime import datetime, timedelta
|
||
|
||
# 添加src目录到路径
|
||
current_dir = Path(__file__).parent
|
||
src_dir = current_dir / "src"
|
||
sys.path.insert(0, str(src_dir))
|
||
|
||
from loguru import logger
|
||
from src.utils.config_loader import config_loader
|
||
from src.data.data_fetcher import ADataFetcher
|
||
from src.utils.notification import NotificationManager
|
||
from src.strategy.kline_pattern_strategy import KLinePatternStrategy
|
||
|
||
|
||
def create_test_signal():
|
||
"""创建一个测试用的K线形态信号"""
|
||
test_signal = {
|
||
'stock_code': '000001.SZ',
|
||
'stock_name': '平安银行',
|
||
'date': datetime.now() - timedelta(days=5), # 5天前的信号
|
||
'timeframe': 'daily',
|
||
'breakout_price': 15.50, # 突破价格
|
||
'yin_high': 15.20, # 阴线最高点
|
||
'pattern_type': '两阳+阴+阳突破'
|
||
}
|
||
return test_signal
|
||
|
||
|
||
def create_test_pullback_data():
|
||
"""创建模拟回踩数据 - 模拟价格回踩到阴线最高点附近"""
|
||
dates = pd.date_range(start=datetime.now() - timedelta(days=3), end=datetime.now(), freq='D')
|
||
|
||
# 模拟价格回踩阴线最高点的情况
|
||
test_data = []
|
||
yin_high = 15.20 # 阴线最高点价格
|
||
initial_price = 15.50 # 突破价格
|
||
|
||
# 设计价格走势：从突破价格逐步回调到阴线最高点附近
|
||
price_path = [15.45, 15.35, 15.25, 15.18] # 最后一个价格接近阴线最高点
|
||
|
||
for i, date in enumerate(dates):
|
||
if i < len(price_path):
|
||
close_price = price_path[i]
|
||
else:
|
||
close_price = yin_high - 0.02 # 稍低于阴线最高点
|
||
|
||
low_price = close_price - 0.03 # 最低价更接近阴线最高点
|
||
|
||
test_data.append({
|
||
'trade_date': date,
|
||
'open': close_price + 0.02,
|
||
'high': close_price + 0.05,
|
||
'low': low_price,
|
||
'close': close_price,
|
||
'volume': 1000000
|
||
})
|
||
|
||
return pd.DataFrame(test_data)
|
||
|
||
|
||
def test_pullback_detection():
|
||
"""测试回踩检测功能"""
|
||
logger.info("🧪 开始测试价格回踩阴线最高点提醒功能")
|
||
|
||
# 初始化配置
|
||
config = config_loader.load_config()
|
||
|
||
# 初始化组件
|
||
data_fetcher = ADataFetcher()
|
||
notification_config = config.get('notification', {})
|
||
notification_manager = NotificationManager(notification_config)
|
||
|
||
# 获取K线形态策略配置
|
||
kline_config = config.get('strategy', {}).get('kline_pattern', {})
|
||
strategy = KLinePatternStrategy(data_fetcher, notification_manager, kline_config)
|
||
|
||
# 创建测试信号
|
||
test_signal = create_test_signal()
|
||
logger.info(f"📊 创建测试信号: {test_signal['stock_code']}({test_signal['stock_name']})")
|
||
logger.info(f" - 信号日期: {test_signal['date']}")
|
||
logger.info(f" - 突破价格: {test_signal['breakout_price']}")
|
||
logger.info(f" - 阴线最高点: {test_signal['yin_high']}")
|
||
|
||
# 添加到策略的监控列表
|
||
strategy.add_triggered_signal(test_signal)
|
||
logger.info("✅ 测试信号已添加到监控列表")
|
||
|
||
# 创建模拟回踩数据
|
||
test_pullback_data = create_test_pullback_data()
|
||
logger.info(f"📈 创建模拟K线数据，共{len(test_pullback_data)}条记录")
|
||
|
||
print("\n模拟K线数据:")
|
||
for _, row in test_pullback_data.iterrows():
|
||
print(f" {row['trade_date'].strftime('%Y-%m-%d')}: "
|
||
f"开{row['open']:.2f} 高{row['high']:.2f} 低{row['low']:.2f} 收{row['close']:.2f}")
|
||
|
||
# 检测回踩情况
|
||
logger.info("🔍 开始检测回踩情况...")
|
||
pullback_alerts = strategy.check_pullback_signals(test_signal['stock_code'], test_pullback_data)
|
||
|
||
if pullback_alerts:
|
||
logger.info(f"⚠️ 检测到 {len(pullback_alerts)} 个回踩提醒")
|
||
for i, alert in enumerate(pullback_alerts, 1):
|
||
logger.info(f" 提醒{i}: {alert['stock_code']} - 当前价格{alert['current_price']:.2f}，"
|
||
f"回调{alert['pullback_pct']:.2f}%，距阴线高点{alert['distance_to_yin_high']:.2f}%")
|
||
|
||
# 测试通知发送（如果启用了钉钉通知）
|
||
if notification_config.get('dingtalk', {}).get('enabled', False):
|
||
logger.info("📱 测试发送回踩提醒通知...")
|
||
success = notification_manager.send_pullback_alerts(pullback_alerts)
|
||
if success:
|
||
logger.info("✅ 回踩提醒通知发送成功")
|
||
else:
|
||
logger.warning("❌ 回踩提醒通知发送失败")
|
||
else:
|
||
logger.info("ℹ️ 钉钉通知未启用，跳过通知发送测试")
|
||
else:
|
||
logger.info("ℹ️ 未检测到回踩情况")
|
||
|
||
# 测试完整的监控流程
|
||
logger.info("\n🔍 测试完整的回踩监控流程...")
|
||
all_pullback_alerts = strategy.monitor_pullback_for_triggered_signals()
|
||
|
||
logger.info("🎯 测试完成!")
|
||
if all_pullback_alerts:
|
||
logger.info(f"✅ 成功检测到 {len(all_pullback_alerts)} 个回踩提醒")
|
||
else:
|
||
logger.info("ℹ️ 当前监控中无回踩情况")
|
||
|
||
|
||
def test_strategy_config():
|
||
"""测试策略配置是否正确加载"""
|
||
logger.info("🔧 测试策略配置加载...")
|
||
|
||
config = config_loader.load_config()
|
||
kline_config = config.get('strategy', {}).get('kline_pattern', {})
|
||
|
||
logger.info("📋 当前K线形态策略配置:")
|
||
logger.info(f" - 启用状态: {kline_config.get('enabled', False)}")
|
||
logger.info(f" - 前两阳线实体比例: {kline_config.get('min_entity_ratio', 0.55)}")
|
||
logger.info(f" - 最后阳线实体比例: {kline_config.get('final_yang_min_ratio', 0.40)}")
|
||
logger.info(f" - 回踩容忍度: {kline_config.get('pullback_tolerance', 0.02)}")
|
||
logger.info(f" - 监控天数: {kline_config.get('monitor_days', 30)}")
|
||
logger.info(f" - 支持时间周期: {kline_config.get('timeframes', ['daily'])}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 设置日志
|
||
logger.remove()
|
||
logger.add(sys.stdout, level="INFO", format="{time:HH:mm:ss} | {level} | {message}")
|
||
|
||
print("=" * 60)
|
||
print("🧪 价格回踩阴线最高点提醒功能测试")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
# 测试配置加载
|
||
test_strategy_config()
|
||
print()
|
||
|
||
# 测试回踩检测功能
|
||
test_pullback_detection()
|
||
|
||
except Exception as e:
|
||
logger.error(f"测试过程中发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
print("\n" + "=" * 60)
|
||
print("🏁 测试结束")
|
||
print("=" * 60) |