""" 缠论分析系统主程序入口 """ import argparse import logging import sys import os from datetime import datetime # 添加项目根目录到Python路径 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 from web.app import run_app # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler('chan_analysis.log', encoding='utf-8') ] ) logger = logging.getLogger(__name__) def run_command_line_analysis(args): """运行命令行分析""" try: logger.info("开始命令行缠论分析...") # 初始化组件 data_fetcher = DataFetcher(args.exchange) data_processor = DataProcessor() # 获取数据 logger.info(f"获取 {args.symbol} {args.timeframe} 数据...") klines = data_fetcher.fetch_klines( symbol=args.symbol, timeframe=args.timeframe, limit=args.limit ) if klines.empty: logger.error("未获取到数据") return # 数据处理 logger.info("清理和验证数据...") if not data_processor.validate_klines(klines): logger.warning("数据验证失败，继续处理...") klines = data_processor.clean_klines(klines) # 缠论分析 logger.info("开始缠论分析...") analyzer = ChanAnalyzer(klines) analysis_summary = analyzer.run_full_analysis(args.fractal_strength) # 输出分析结果 print("\n" + "="*50) print("缠论分析结果摘要") print("="*50) # 数据信息 data_info = analysis_summary['data_info'] print(f"\n数据信息:") print(f" 原始K线数量: {data_info['original_klines']}") print(f" 处理后K线数量: {data_info['processed_klines']}") print(f" 数据时间范围: {data_info['date_range']['start']} 至 {data_info['date_range']['end']}") # 分型信息 fractal_info = analysis_summary['fractal_info'] print(f"\n分型信息:") print(f" 总分型数量: {fractal_info['total']}") print(f" 顶分型: {fractal_info['top']}") print(f" 底分型: {fractal_info['bottom']}") # 笔信息 stroke_info = analysis_summary['stroke_info'] print(f"\n笔信息:") print(f" 总笔数量: {stroke_info['total']}") print(f" 上升笔: {stroke_info['up']}") print(f" 下降笔: {stroke_info['down']}") # 线段信息 segment_info = analysis_summary['segment_info'] print(f"\n线段信息:") print(f" 总线段数量: {segment_info['total']}") print(f" 上升线段: {segment_info['up']}") print(f" 下降线段: {segment_info['down']}") # 中枢信息 cb_info = analysis_summary['central_bank_info'] print(f"\n中枢信息:") print(f" 总中枢数量: {cb_info['total']}") if cb_info['levels']: print(" 级别分布:") for level, count in cb_info['levels'].items(): print(f" {level}: {count}个") # 买卖点信息 signal_info = analysis_summary['trading_signal_info'] print(f"\n买卖点信息:") print(f" 总信号数量: {signal_info['total']}") print(f" 买点: {signal_info['buy_points']}") print(f" 卖点: {signal_info['sell_points']}") if signal_info['by_class']: print(" 类别分布:") for class_name, count in signal_info['by_class'].items(): print(f" {class_name}: {count}个") # 获取最新信号 latest_signals = analyzer.get_latest_signals(24) # 最近24小时 if latest_signals: print(f"\n最近24小时信号 ({len(latest_signals)}个):") for signal in latest_signals: signal_type = "买点" if signal.signal_type == 'buy' else "卖点" class_type = {"first": "一类", "second": "二类", "third": "三类"}.get(signal.point_class, "未知") print(f" {signal.timestamp.strftime('%m-%d %H:%M')} - {class_type}{signal_type} - 价格:{signal.price:.2f} - 强度:{signal.strength:.3f}") # 当前市场结构 market_structure = analyzer.get_current_market_structure() if market_structure: print(f"\n当前市场结构:") print(f" 当前价格: {market_structure.get('current_price', 'N/A')}") trend_map = {'upward': '上升', 'downward': '下降', 'sideways': '震荡', 'unclear': '不明'} print(f" 趋势方向: {trend_map.get(market_structure.get('trend'), '未知')}") phase_map = {'consolidation': '盘整', 'trending': '趋势', 'unknown': '未知'} print(f" 市场阶段: {phase_map.get(market_structure.get('market_phase'), '未知')}") # 保存结果 if args.export: export_results(analyzer, args) logger.info("命令行分析完成") except Exception as e: logger.error(f"命令行分析失败: {e}") raise def export_results(analyzer: ChanAnalyzer, args): """导出分析结果""" try: results = analyzer.export_results() timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") export_dir = f"exports/chan_analysis_{args.symbol.replace('/', '_')}_{timestamp}" os.makedirs(export_dir, exist_ok=True) # 保存各种分析结果 for name, df in results.items(): if not df.empty: filename = f"{export_dir}/{name}.csv" df.to_csv(filename, encoding='utf-8-sig') logger.info(f"导出 {name} 到 {filename}") # 保存分析摘要 with open(f"{export_dir}/summary.txt", 'w', encoding='utf-8') as f: f.write(f"缠论分析结果摘要\n") f.write(f"交易对: {args.symbol}\n") f.write(f"时间周期: {args.timeframe}\n") f.write(f"分析时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"数据数量: {args.limit}\n") f.write(f"分型强度: {args.fractal_strength}\n") logger.info(f"分析结果已导出到 {export_dir}") except Exception as e: logger.error(f"导出结果失败: {e}") def main(): """主程序入口""" parser = argparse.ArgumentParser(description='缠论分析系统') # 运行模式 parser.add_argument('--mode', choices=['web', 'cli'], default='web', help='运行模式：web(Web界面) 或 cli(命令行)') # 数据参数 parser.add_argument('--symbol', default='BTC/USDT', help='交易对 (默认: BTC/USDT)') parser.add_argument('--timeframe', default='1h', help='时间周期 (默认: 1h)') parser.add_argument('--limit', type=int, default=500, help='获取的K线数量 (默认: 500)') parser.add_argument('--exchange', default='binance', help='交易所 (默认: binance)') # 分析参数 parser.add_argument('--fractal-strength', type=int, default=1, help='分型强度要求 (默认: 1)') # 输出参数 parser.add_argument('--export', action='store_true', help='是否导出分析结果') # Web参数 parser.add_argument('--port', type=int, default=8050, help='Web服务端口 (默认: 8050)') parser.add_argument('--debug', action='store_true', help='是否启用调试模式') args = parser.parse_args() logger.info("缠论分析系统启动") logger.info(f"运行模式: {args.mode}") try: if args.mode == 'web': logger.info(f"启动Web服务，端口: {args.port}") print(f"\n缠论分析系统Web界面") print(f"访问地址: http://localhost:{args.port}") print("按 Ctrl+C 停止服务\n") run_app(debug=args.debug, port=args.port) else: run_command_line_analysis(args) except KeyboardInterrupt: logger.info("用户中断，程序退出") except Exception as e: logger.error(f"程序执行失败: {e}") sys.exit(1) if __name__ == "__main__": main()