166 lines
5.2 KiB
Python
166 lines
5.2 KiB
Python
"""应用配置管理"""
|
||
|
||
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 = "75981230@qq.com"
|
||
admin_email: str = "75981230@qq.com"
|
||
admin_password: str = "880803"
|
||
|
||
# 认证体系
|
||
auth_min_password_length: int = 6
|
||
invite_code_required: bool = True
|
||
email_code_expiry_minutes: int = 10
|
||
email_code_cooldown_seconds: int = 60
|
||
|
||
# SMTP
|
||
smtp_host: str = ""
|
||
smtp_port: int = 465
|
||
smtp_username: str = ""
|
||
smtp_password: str = ""
|
||
smtp_sender: str = ""
|
||
smtp_use_ssl: bool = True
|
||
|
||
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. 交易日 09:15 后都优先实时源，包括收盘后；腾讯/东方财富会保留当日收盘快照
|
||
3. 盘前不使用今日实时源，避免拿到昨日收盘价却标成今日
|
||
"""
|
||
from zoneinfo import ZoneInfo
|
||
|
||
now = datetime.now(ZoneInfo("Asia/Shanghai"))
|
||
if now.weekday() >= 5:
|
||
return False
|
||
|
||
t = now.hour * 100 + now.minute
|
||
if t < 915:
|
||
return False
|
||
|
||
today = today_trade_date()
|
||
if latest_trade_date and latest_trade_date.replace("-", "") > today:
|
||
return False
|
||
|
||
return True
|