"""应用配置管理""" import os from datetime import datetime from pydantic_settings import BaseSettings class Settings(BaseSettings): # Tushare Pro tushare_token: str = "" # 服务配置 host: str = "0.0.0.0" port: int = 8000 debug: bool = False # 数据库 database_url: str = "sqlite:///./astock.db" # 缓存 TTL(秒) cache_ttl_realtime: int = 30 # 实时数据 30 秒 cache_ttl_daily: int = 300 # 日级数据 5 分钟 cache_ttl_sector: int = 300 # 板块数据 5 分钟 cache_ttl_static: int = 86400 # 静态数据 24 小时 # Tushare 限流 tushare_request_delay: float = 0.3 # 请求间隔(秒) tushare_max_retry: int = 3 # 最大重试次数 # 筛选参数 top_sector_count: int = 5 # 关注板块数量 top_stock_count: int = 20 # 进入技术面筛选的个股数 candidate_pool_limit: int = 120 # 多路召回后的候选池上限 llm_prefilter_limit: int = 36 # LLM 初筛保留数量 llm_prefilter_max_concurrent: int = 6 llm_final_limit: int = 14 # LLM 深裁决池上限 min_turnover_rate: float = 2.0 # 最小换手率 % max_turnover_rate: float = 30.0 # 最大换手率 % min_circ_mv: float = 50.0 # 最小流通市值(亿) max_circ_mv: float = 500.0 # 最大流通市值(亿) min_list_days: int = 60 # 最小上市天数 # 买入信号阈值 buy_score_threshold: int = 60 # 买入最低分 buy_min_signals: int = 3 # 最少满足信号数 # 风控 stop_loss_pct: float = 5.0 # 止损比例 % # 趋势突破策略参数 breakout_min_volume_ratio: float = 1.2 # 突破型最小量比 pullback_max_shrink_ratio: float = 0.85 # 回踩型最大缩量比 consolidation_max_range_pct: float = 8.0 # 启动型最大整理振幅 % # LLM (DeepSeek) deepseek_api_key: str = "" deepseek_base_url: str = "https://api.deepseek.com/v1" deepseek_model: str = "deepseek-chat" llm_max_tokens: int = 2000 llm_temperature: float = 0.3 # 告警(Feishu / Lark Incoming Webhook) alert_enabled: bool = False feishu_webhook_url: str = "" alert_dedup_ttl_seconds: int = 300 alert_max_detail_chars: int = 1200 alert_app_name: str = "AStock Agent" alert_environment: str = "local" # 前端 frontend_url: str = "http://localhost:3002" # JWT 认证 jwt_secret: str = "change-me-in-production" jwt_expiry_hours: int = 24 jwt_algorithm: str = "HS256" # 默认管理员(首次启动自动创建) admin_username: str = "admin" admin_password: str = "admin123" model_config = {"env_file": ".env", "env_prefix": "ASTOCK_"} settings = Settings() def is_trading_hours() -> bool: """判断当前是否在 A 股交易时段(9:30-11:30, 13:00-15:00)""" from zoneinfo import ZoneInfo now = datetime.now(ZoneInfo("Asia/Shanghai")) weekday = now.weekday() # 0=Mon, 6=Sun if weekday >= 5: return False t = now.hour * 100 + now.minute return (930 <= t <= 1130) or (1300 <= t <= 1500) def is_market_session() -> bool: """判断当前是否在 A 股交易日内(含午休 11:30-13:00) 午休期间腾讯实时行情仍返回上午收盘价,可用于展示。 与 is_trading_hours() 的区别:午休时返回 True。 """ from zoneinfo import ZoneInfo now = datetime.now(ZoneInfo("Asia/Shanghai")) weekday = now.weekday() if weekday >= 5: return False t = now.hour * 100 + now.minute return 930 <= t <= 1500 def is_pre_close() -> bool: """判断是否在收盘后、数据更新前(15:00-15:30 数据尚未完全更新)""" from zoneinfo import ZoneInfo now = datetime.now(ZoneInfo("Asia/Shanghai")) t = now.hour * 100 + now.minute return 1500 <= t <= 1530 def today_trade_date() -> str: """返回上海时区下的今天日期(YYYYMMDD)。""" from zoneinfo import ZoneInfo return datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d") def should_prefer_realtime_today(latest_trade_date: str | None = None) -> bool: """是否应该优先使用“今天”的实时数据。 规则: 1. 非交易日直接 False 2. 交易日内(含午休、收盘后)如果 Tushare 最新交易日还不是今天,则优先实时 3. 即便 Tushare 最新交易日已经是今天,只要仍处于 15:30 前的延迟窗口,也优先实时 """ if not is_market_session() and not is_pre_close(): return False today = today_trade_date() if latest_trade_date and latest_trade_date != today: return True return is_market_session() or is_pre_close()