Rekey
039bfb5075
- engine/__init__.py: 包入口，导出 Kline/KlineInterval/OrderBook/Ticker/Trade - common/base.py: BaseStrategy 抽象基类，定义 on_kline/on_ticker/on_orderbook 回调 - common/models.py: Pydantic 数据模型，与 TS 侧 types 字段对齐，支持字段校验 - common/config.py: 全局配置加载（YAML），统一 engine/env.yaml 读取 - common/logger.py: 结构化日志，支持 JSON/pretty print 输出
73 lines
1.5 KiB
Python
73 lines
1.5 KiB
Python
"""
|
|
项目配置模块 — 读取并校验根目录 env.yaml
|
|
|
|
使用方式：
|
|
from engine.common.config import config
|
|
|
|
db = config.db
|
|
print(db.host, db.port, db.name)
|
|
print(config.redis.url)
|
|
print(config.logging.level)
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import yaml
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class DBConfig(BaseModel):
|
|
"""TimescaleDB / PostgreSQL 连接配置"""
|
|
|
|
host: str
|
|
port: int = 5432
|
|
name: str
|
|
user: str
|
|
password: str
|
|
|
|
|
|
class RedisConfig(BaseModel):
|
|
"""Redis 连接配置"""
|
|
|
|
url: str = "redis://localhost:6379"
|
|
publish_enabled: bool = True
|
|
|
|
|
|
class LoggingConfig(BaseModel):
|
|
"""日志配置"""
|
|
|
|
level: str = "debug" # trace / debug / info / warn / error / fatal
|
|
node_env: str = Field(default="development", alias="node_env")
|
|
|
|
class Config:
|
|
populate_by_name = True
|
|
|
|
|
|
class AppConfig(BaseModel):
|
|
"""应用配置聚合"""
|
|
|
|
db: DBConfig
|
|
redis: RedisConfig
|
|
logging: LoggingConfig
|
|
|
|
|
|
def load_config(config_path: Optional[Path] = None) -> AppConfig:
|
|
"""从 env.yaml 加载并校验配置
|
|
|
|
Args:
|
|
config_path: 显式指定路径；为 None 时自动查找项目根目录 env.yaml
|
|
"""
|
|
if config_path is None:
|
|
# engine/common/config.py → engine/ (parent of common/) → 根目录 env.yaml
|
|
config_path = Path(__file__).resolve().parent.parent / "env.yaml"
|
|
|
|
with open(config_path) as f:
|
|
raw = yaml.safe_load(f)
|
|
|
|
return AppConfig(**raw)
|
|
|
|
|
|
# ── 模块级单例 ──
|
|
config = load_config()
|