Function run

cli/main.py:908–990

运行ContestTrade分析

(
    market: Optional[str] = typer.Option(None, "--market", "-m", help="选择市场 (CN-Stock/US-Stock)"),
)

@app.command()
def run(
    market: Optional[str] = typer.Option(None, "--market", "-m", help="选择市场 (CN-Stock/US-Stock)"),
):
    """运行ContestTrade分析"""

    # 获取市场选择
    if not market:
        market = get_market_selection()
    
    # 验证市场选择
    if not market:
        console.print("[red]未提供市场选择[/red]")
        raise typer.Exit(1)
    
    if market not in ["CN-Stock", "US-Stock"]:
        console.print("[red]市场选择错误，请选择 CN-Stock 或 US-Stock[/red]")
        raise typer.Exit(1)
    
    # 设置环境变量 - 这样全局的 cfg 就会读取对应的配置
    os.environ['CONTEST_TRADE_MARKET'] = market
    
    # 根据市场获取对应的触发时间
    trigger_time = get_trigger_time_for_market(market)
    
    # 验证触发时间
    if not trigger_time:
        console.print("[red]无法获取对应市场的触发时间[/red]")
        raise typer.Exit(1)
    
    console.print(f"[green]已选择市场: {market}[/green]")
    console.print(f"[green]触发时间: {trigger_time}[/green]")
    
    # 验证必需的服务连接
    if not validate_required_services():
        console.print("[red]系统验证失败，无法启动分析[/red]")
        raise typer.Exit(1)
    
    # 主循环
    while True:
        try:
            result = run_contest_analysis_interactive(trigger_time, market)
        except Exception as e:
            console.print(f"[red]运行分析时发生错误: {e}[/red]")
            break
        
        if result is None or (isinstance(result, tuple) and result[0] is None):
            console.print("[red]❌ 分析失败[/red]")
            break
            
        if isinstance(result, tuple):
            final_state, action = result
            if action == "new_analysis":
                # 重新选择市场
                market = get_market_selection()
                if not market:
                    break
                
                # 设置环境变量