72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
|
"""全局配置：从环境变量 / .env 读取。
|
|||
|
|
|
|||
|
|
关键点：
|
|||
|
|
- OWN_DB_DSN 自有 Postgres（看板只读它，ETL 写它）—— 必填
|
|||
|
|
- LEGACY_MYSQL_DSN 现有 MySQL-A(18.199)，读 zs_day_data —— 选填，留空则跳过指数同步
|
|||
|
|
- LEGACY_PG_DSN 现有 PG(16.150)，读 gp_market_sentiment —— 选填，留空则跳过情绪同步
|
|||
|
|
- OHLC_PROVIDER 指数 OHLC 数据源（你后续提供）。默认 "none" → 退化为收盘线
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from functools import lru_cache
|
|||
|
|
from typing import List, Optional
|
|||
|
|
|
|||
|
|
from pydantic import Field
|
|||
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Settings(BaseSettings):
|
|||
|
|
model_config = SettingsConfigDict(
|
|||
|
|
env_file=".env", env_file_encoding="utf-8", extra="ignore"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# ---- 数据库 ----
|
|||
|
|
own_db_dsn: str = Field(
|
|||
|
|
default="postgresql+psycopg://asevent:asevent@db:5432/asevent",
|
|||
|
|
alias="OWN_DB_DSN",
|
|||
|
|
)
|
|||
|
|
legacy_mysql_dsn: Optional[str] = Field(default=None, alias="LEGACY_MYSQL_DSN")
|
|||
|
|
legacy_pg_dsn: Optional[str] = Field(default=None, alias="LEGACY_PG_DSN")
|
|||
|
|
|
|||
|
|
# ---- 指数范围（dot 式，与 zs_day_data.symbol 一致）----
|
|||
|
|
index_codes: str = Field(default="000001.SH,399006.SZ,000688.SH", alias="INDEX_CODES")
|
|||
|
|
|
|||
|
|
# ---- OHLC 源（预留插槽）----
|
|||
|
|
# "none" = 无源，open=high=low=close，ohlc_source='close_only'（退化为收盘线）
|
|||
|
|
# 你接入后在 sources.py:build_ohlc_provider 里注册自己的实现，并把此值改成对应名字
|
|||
|
|
ohlc_provider: str = Field(default="none", alias="OHLC_PROVIDER")
|
|||
|
|
|
|||
|
|
# ---- 顶底信号参数（初值，实机回测后再调）----
|
|||
|
|
vol_window: int = Field(default=250, alias="VOL_WINDOW") # 量能/价格分位滚动窗口
|
|||
|
|
w_vol: float = Field(default=0.35, alias="W_VOL")
|
|||
|
|
w_price: float = Field(default=0.25, alias="W_PRICE")
|
|||
|
|
w_sentiment: float = Field(default=0.25, alias="W_SENTIMENT")
|
|||
|
|
w_etf: float = Field(default=0.15, alias="W_ETF")
|
|||
|
|
hot_threshold: float = Field(default=80.0, alias="HOT_THRESHOLD") # 过热/顶部
|
|||
|
|
cold_threshold: float = Field(default=20.0, alias="COLD_THRESHOLD") # 冰点/底部
|
|||
|
|
|
|||
|
|
# ---- 其它 ----
|
|||
|
|
cors_origins: str = Field(default="*", alias="CORS_ORIGINS")
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def index_code_list(self) -> List[str]:
|
|||
|
|
return [c.strip() for c in self.index_codes.split(",") if c.strip()]
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def cors_origin_list(self) -> List[str]:
|
|||
|
|
return [c.strip() for c in self.cors_origins.split(",") if c.strip()]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@lru_cache
|
|||
|
|
def get_settings() -> Settings:
|
|||
|
|
return Settings()
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 指数中文名映射（展示用）
|
|||
|
|
INDEX_NAMES = {
|
|||
|
|
"000001.SH": "上证综指",
|
|||
|
|
"399006.SZ": "创业板指",
|
|||
|
|
"000688.SH": "科创50",
|
|||
|
|
"399001.SZ": "深证成指",
|
|||
|
|
}
|