213 lines
7.2 KiB
Python
213 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
缠论分析系统快速启动脚本
|
|
用于快速演示系统功能
|
|
"""
|
|
|
|
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.abspath(__file__)))
|
|
|
|
from data.data_fetcher import DataFetcher
|
|
from data.data_processor import DataProcessor
|
|
from core.chan_analyzer import ChanAnalyzer
|
|
|
|
|
|
def quick_demo():
|
|
"""快速演示"""
|
|
print("=" * 60)
|
|
print("缠论分析系统快速演示")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
print("正在初始化组件...")
|
|
|
|
# 初始化数据获取器
|
|
data_fetcher = DataFetcher()
|
|
data_processor = DataProcessor()
|
|
|
|
print("正在获取BTC/USDT 1小时数据...")
|
|
|
|
# 获取数据
|
|
klines = data_fetcher.fetch_klines('BTC/USDT', '1h', 200)
|
|
|
|
if klines.empty:
|
|
print("❌ 数据获取失败，可能是网络问题")
|
|
return False
|
|
|
|
print(f"✅ 成功获取 {len(klines)} 条K线数据")
|
|
print(f"数据时间范围: {klines.index[0]} 至 {klines.index[-1]}")
|
|
|
|
# 数据处理
|
|
print("\n正在进行数据清理和验证...")
|
|
klines = data_processor.clean_klines(klines)
|
|
|
|
# 缠论分析
|
|
print("正在进行缠论分析...")
|
|
analyzer = ChanAnalyzer(klines)
|
|
summary = analyzer.run_full_analysis(fractal_strength=1)
|
|
|
|
# 显示结果
|
|
print("\n" + "=" * 60)
|
|
print("缠论分析结果")
|
|
print("=" * 60)
|
|
|
|
# 基础统计
|
|
data_info = summary['data_info']
|
|
print(f"\n📊 数据统计:")
|
|
print(f" 原始K线: {data_info['original_klines']} 条")
|
|
print(f" 处理后: {data_info['processed_klines']} 条")
|
|
|
|
# 分型统计
|
|
fractal_info = summary['fractal_info']
|
|
print(f"\n🔺 分型统计:")
|
|
print(f" 总计: {fractal_info['total']} 个")
|
|
print(f" 顶分型: {fractal_info['top']} 个")
|
|
print(f" 底分型: {fractal_info['bottom']} 个")
|
|
|
|
# 笔统计
|
|
stroke_info = summary['stroke_info']
|
|
print(f"\n📏 笔统计:")
|
|
print(f" 总计: {stroke_info['total']} 个")
|
|
print(f" 上升笔: {stroke_info['up']} 个")
|
|
print(f" 下降笔: {stroke_info['down']} 个")
|
|
|
|
# 线段统计
|
|
segment_info = summary['segment_info']
|
|
print(f"\n📈 线段统计:")
|
|
print(f" 总计: {segment_info['total']} 个")
|
|
print(f" 上升线段: {segment_info['up']} 个")
|
|
print(f" 下降线段: {segment_info['down']} 个")
|
|
|
|
# 中枢统计
|
|
cb_info = summary['central_bank_info']
|
|
print(f"\n🎯 中枢统计:")
|
|
print(f" 总计: {cb_info['total']} 个")
|
|
if cb_info['levels']:
|
|
for level, count in cb_info['levels'].items():
|
|
print(f" {level}: {count} 个")
|
|
|
|
# 买卖点统计
|
|
signal_info = summary['trading_signal_info']
|
|
print(f"\n💰 买卖点统计:")
|
|
print(f" 总计: {signal_info['total']} 个")
|
|
print(f" 买点: {signal_info['buy_points']} 个")
|
|
print(f" 卖点: {signal_info['sell_points']} 个")
|
|
|
|
# 最新信号
|
|
latest_signals = analyzer.get_latest_signals(48) # 最近48小时
|
|
if latest_signals:
|
|
print(f"\n🚨 最近48小时信号 ({len(latest_signals)} 个):")
|
|
for signal in latest_signals[-5:]: # 最多显示5个
|
|
signal_type = "🟢买点" if signal.signal_type == 'buy' else "🔴卖点"
|
|
class_map = {"first": "一类", "second": "二类", "third": "三类"}
|
|
class_type = class_map.get(signal.point_class, "未知")
|
|
time_str = signal.timestamp.strftime('%m-%d %H:%M')
|
|
print(f" {time_str} | {class_type}{signal_type} | 价格: ${signal.price:,.2f} | 强度: {signal.strength:.3f}")
|
|
|
|
# 当前市场结构
|
|
market_structure = analyzer.get_current_market_structure()
|
|
if market_structure:
|
|
print(f"\n📊 当前市场结构:")
|
|
current_price = market_structure.get('current_price', 0)
|
|
print(f" 当前价格: ${current_price:,.2f}")
|
|
|
|
trend_map = {
|
|
'upward': '🟢 上升趋势',
|
|
'downward': '🔴 下降趋势',
|
|
'sideways': '🟡 震荡趋势',
|
|
'unclear': '⚪ 趋势不明'
|
|
}
|
|
trend = trend_map.get(market_structure.get('trend'), '未知')
|
|
print(f" 趋势方向: {trend}")
|
|
|
|
phase_map = {
|
|
'consolidation': '📦 盘整阶段',
|
|
'trending': '📈 趋势阶段',
|
|
'unknown': '❓ 未知阶段'
|
|
}
|
|
phase = phase_map.get(market_structure.get('market_phase'), '未知')
|
|
print(f" 市场阶段: {phase}")
|
|
|
|
# 支撑阻力
|
|
sr = market_structure.get('support_resistance', {})
|
|
if sr.get('support_levels'):
|
|
supports = [f"${s:,.2f}" for s in sr['support_levels'][:3]]
|
|
print(f" 关键支撑: {', '.join(supports)}")
|
|
if sr.get('resistance_levels'):
|
|
resistances = [f"${r:,.2f}" for r in sr['resistance_levels'][:3]]
|
|
print(f" 关键阻力: {', '.join(resistances)}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("分析完成！")
|
|
print("=" * 60)
|
|
|
|
# 启动Web界面提示
|
|
print(f"\n💡 想要查看可视化图表？运行以下命令启动Web界面：")
|
|
print(f" python main.py --mode web")
|
|
print(f" 然后在浏览器中访问 http://localhost:8050")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ 演示失败: {e}")
|
|
print("可能的原因：")
|
|
print("1. 网络连接问题，无法获取数据")
|
|
print("2. 缺少必要的依赖包")
|
|
print("3. 交易所API限制")
|
|
print("\n解决方案：")
|
|
print("1. 检查网络连接")
|
|
print("2. 运行: pip install -r requirements.txt")
|
|
print("3. 稍后重试")
|
|
return False
|
|
|
|
|
|
def check_dependencies():
|
|
"""检查依赖包"""
|
|
required_packages = [
|
|
'ccxt', 'pandas', 'numpy', 'plotly', 'dash'
|
|
]
|
|
|
|
missing_packages = []
|
|
|
|
for package in required_packages:
|
|
try:
|
|
__import__(package)
|
|
except ImportError:
|
|
missing_packages.append(package)
|
|
|
|
if missing_packages:
|
|
print("❌ 缺少以下依赖包:")
|
|
for pkg in missing_packages:
|
|
print(f" - {pkg}")
|
|
print("\n请运行以下命令安装:")
|
|
print("pip install -r requirements.txt")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("正在检查依赖包...")
|
|
|
|
if not check_dependencies():
|
|
sys.exit(1)
|
|
|
|
print("✅ 依赖包检查通过")
|
|
|
|
if quick_demo():
|
|
print("\n🎉 演示成功完成！")
|
|
sys.exit(0)
|
|
else:
|
|
print("\n❌ 演示失败")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |