feat(backend): 搭建 FastAPI 分层骨架 — 配置加载 / 健康检查 / SQLAlchemy+Alembic 就位

- uv 工程(Python 3.12,pyproject + uv.lock,dev: pytest/ruff)
- 分层包结构:api / application / domain / infrastructure / quant(qlib_adapter) / agent / core
- app.core.config:加载根 config.yaml + .env,sqlite 相对路径解析到项目根 data/
- SQLAlchemy Base + Session 工厂(Infrastructure 层,业务走 Repository)
- Alembic 迁移环境就位(env.py 复用应用配置,render_as_batch 兼容 SQLite)
- GET /api/health + pytest 冒烟 6 项通过 / ruff clean / uvicorn 启动验证
This commit is contained in:
Simon
2026-09-06 16:09:10 +08:00
parent 8eb3b4ac2a
commit 2a52ee5e83
32 changed files with 1443 additions and 0 deletions
+3
View File
@@ -0,0 +1,3 @@
"""qlib-platform backend 应用包。"""
__version__ = "0.1.0"
+6
View File
@@ -0,0 +1,6 @@
"""AI Research Agent(Phase 5 接入)。
Agent 是 Research Assistant 而非系统管理员:只能通过受控 Tool
(search_stock / test_factor / run_backtest / get_experiment ...)访问研究能力,
禁止直接操作数据库 / shell / 生产配置。
"""
+1
View File
@@ -0,0 +1 @@
"""Agent 受控工具集:每个 Tool 是后端服务的只读 / 沙箱化入口。"""
+7
View File
@@ -0,0 +1,7 @@
"""HTTP API 层。
约束(AGENT.md §17/§18):
- 路由面向业务对象(/api/stocks、/api/factors、/api/backtests ...),不暴露 Qlib / SQL
- 输入输出一律 Pydantic DTO,禁止把 ORM Model 直接暴露
- 耗时任务返回 job_id,不长时间阻塞 HTTP
"""
+19
View File
@@ -0,0 +1,19 @@
"""系统健康检查。"""
from __future__ import annotations
from fastapi import APIRouter
from app.core.config import get_settings
router = APIRouter(tags=["system"])
@router.get("/health", summary="健康检查")
def health() -> dict:
settings = get_settings()
return {
"status": "ok",
"app": settings.app_name,
"version": settings.app_version,
}
+14
View File
@@ -0,0 +1,14 @@
"""API 路由聚合。
后续业务路由按 AGENT.md §17 面向业务对象挂载:
/api/stocks /api/universes /api/factors /api/strategies /api/backtests /api/experiments /api/jobs
"""
from __future__ import annotations
from fastapi import APIRouter
from app.api import health
api_router = APIRouter()
api_router.include_router(health.router)
+5
View File
@@ -0,0 +1,5 @@
"""应用服务层:用例编排(Data / Factor / Strategy / Backtest / Experiment ...)。
约束:本层只依赖 domain Repository Protocol 与 quant 服务,禁止出现 sqlite3 / SQL /
直接 import Qlib 内部实现。
"""
@@ -0,0 +1 @@
"""用例 / 应用服务实现。每个服务对应一个用例,保持小而可测。"""
+1
View File
@@ -0,0 +1 @@
"""核心组件(配置、通用工具)。"""
+126
View File
@@ -0,0 +1,126 @@
"""核心配置层:加载根目录 config.yaml + .env,向全应用提供 Settings。
加载优先级:环境变量 > 根目录 .env > 根目录 config.yaml > 代码默认值。
密钥只存在于 .env(config.yaml 经 *_env 字段引用变量名)。
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
import yaml
# 项目根:<root>/backend/app/core/config.py → parents[3]
PROJECT_ROOT = Path(__file__).resolve().parents[3]
BACKEND_ROOT = PROJECT_ROOT / "backend"
CONFIG_PATH = PROJECT_ROOT / "config.yaml"
ENV_PATH = PROJECT_ROOT / ".env"
_SQLITE_PREFIX = "sqlite:///"
def _load_env_file(path: Path) -> None:
"""把 .env 读入 os.environ(不覆盖已存在的环境变量)。"""
if not path.exists():
return
for raw in path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
key = key.strip()
if key and key not in os.environ:
os.environ[key] = value.strip().strip("\"'")
def _load_config_yaml(path: Path) -> dict:
if not path.exists():
return {}
with path.open(encoding="utf-8") as fh:
data = yaml.safe_load(fh)
return data if isinstance(data, dict) else {}
def _deep(cfg: dict, dotted: str) -> object:
node: object = cfg
for part in dotted.split("."):
if not isinstance(node, dict) or part not in node:
return None
node = node[part]
return node
@dataclass(frozen=True)
class Settings:
"""应用全局配置(已解析、不可变)。"""
app_name: str
app_version: str
debug: bool
secret_key: str
api_prefix: str
database_url: str
sqlalchemy_echo: bool
migrations_dir: Path
data_source_primary: str
data_source_fallback: str
tushare_token: str
storage: dict[str, Path]
config_path: Path = CONFIG_PATH
env_path: Path = ENV_PATH
project_root: Path = PROJECT_ROOT
def _resolve_storage_dirs(cfg: dict) -> dict[str, Path]:
storage_cfg = _deep(cfg, "storage") or {}
return {
key: PROJECT_ROOT / str(value)
for key, value in storage_cfg.items()
if isinstance(value, str)
}
def _normalize_sqlite_url(url: str) -> str:
"""把相对路径的 sqlite URL 解析为项目根下的绝对路径。"""
if not url.startswith(_SQLITE_PREFIX):
return url
rest = url[len(_SQLITE_PREFIX) :]
if rest.startswith("/"): # 已是绝对路径(sqlite:////abs/path)
return url
return f"{_SQLITE_PREFIX}{(PROJECT_ROOT / rest).resolve()}"
@lru_cache
def get_settings() -> Settings:
"""加载并缓存 Settings。config / env 路径可通过参数覆盖以便测试。"""
_load_env_file(ENV_PATH)
cfg = _load_config_yaml(CONFIG_PATH)
secret_env = _deep(cfg, "app.secret_key_env") or "APP_SECRET_KEY"
url_env = _deep(cfg, "database.url_env") or "DATABASE_URL"
token_env = _deep(cfg, "data_source.tushare_token_env") or "TUSHARE_TOKEN"
default_url = "sqlite:///./data/quant.db"
database_url = os.environ.get(url_env) or default_url
migrations_rel = _deep(cfg, "database.migrations_dir") or (
"app/infrastructure/persistence/migrations"
)
return Settings(
app_name=str(_deep(cfg, "app.name") or "qlib-platform"),
app_version=str(_deep(cfg, "app.version") or "0.1.0"),
debug=bool(_deep(cfg, "app.debug") or False),
secret_key=os.environ.get(secret_env, "dev-insecure-key"),
api_prefix=str(_deep(cfg, "api.prefix") or "/api"),
database_url=_normalize_sqlite_url(database_url),
sqlalchemy_echo=bool(_deep(cfg, "database.echo") or False),
migrations_dir=BACKEND_ROOT / str(migrations_rel),
data_source_primary=str(_deep(cfg, "data_source.primary") or "tushare"),
data_source_fallback=str(_deep(cfg, "data_source.fallback") or "sina"),
tushare_token=os.environ.get(token_env, ""),
storage=_resolve_storage_dirs(cfg),
)
+5
View File
@@ -0,0 +1,5 @@
"""领域层:纯业务实体与规则,不依赖框架 / 数据库。
约束(AGENT.md §10):业务层禁止操作 sqlite3 / SQL / Session,只允许调用
domain/repositories 中定义的 Repository Protocol。
"""
+4
View File
@@ -0,0 +1,4 @@
"""领域实体(Stock / Universe / Factor / Strategy / Backtest / Experiment / Job ...)。
Phase 1 逐步补充,见 docs/ARCHITECTURE.md §10。
"""
@@ -0,0 +1,11 @@
"""Repository Protocol(接口)。
业务层只依赖这里定义的 Protocol,例如:
class StockRepository(Protocol):
def get_by_symbol(self, symbol: str) -> Stock | None: ...
def list(self, filters: StockFilter) -> list[Stock]: ...
def save(self, stock: Stock) -> Stock: ...
具体实现放在 infrastructure/persistence 下(SQLite 当前实现,未来 MySQL 同接口)。
"""
+1
View File
@@ -0,0 +1 @@
"""基础设施层:数据库实现(SQLAlchemy / Alembic)等外部依赖的落地实现。"""
@@ -0,0 +1,5 @@
"""持久化子层:SQLAlchemy models / repositories / migrations。
约定:只在这里出现数据库实现细节;业务层不感知底层是 SQLite 还是 MySQL
(通过 DATABASE_URL 切换,见 app.core.config)。
"""
@@ -0,0 +1,64 @@
"""Alembic 迁移环境:从应用配置读取 DATABASE_URL 并连接 Base.metadata。
用法(cd backend):
uv run alembic revision --autogenerate -m "add xxx table"
uv run alembic upgrade head
"""
from __future__ import annotations
from logging.config import fileConfig
from alembic import context
from app.core.config import get_settings
from app.infrastructure.persistence.sqlalchemy.base import Base
from sqlalchemy import engine_from_config, pool
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
config.set_main_option("sqlalchemy.url", get_settings().database_url)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
"""离线模式:只生成 SQL,不连接数据库。"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
render_as_batch=True, # SQLite 支持 ALTER 需要 batch 模式
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""在线模式:连接数据库执行迁移。"""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=True,
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
@@ -0,0 +1,27 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
${imports if imports else ""}
revision: str = ${repr(up_revision)}
down_revision: str | None = ${repr(down_revision)}
branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
depends_on: str | Sequence[str] | None = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}
@@ -0,0 +1 @@
"""SQLAlchemy 2.x 基础设施(当前 SQLite,未来 MySQL 同实现,仅 URL 不同)。"""
@@ -0,0 +1,12 @@
"""所有 ORM Model 的声明式基类。
新表 Model 必须继承本类;改动后通过 Alembic autogenerate 生成迁移。
"""
from __future__ import annotations
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
"""声明式 ORM 基类(SQLAlchemy 2.x 风格)。"""
@@ -0,0 +1,5 @@
"""ORM 模型目录。
新增表流程(AGENT.md §12):Model → Alembic Migration → Test。
模型统一继承 infra.persistence.sqlalchemy.base.Base。
"""
@@ -0,0 +1,5 @@
"""Repository 的 SQLAlchemy 实现。
职责:把 domain Repository Protocol 翻译成 SQLAlchemy 查询。
业务 Service 只依赖 Protocol,不 import 本目录实现(由依赖注入装配)。
"""
@@ -0,0 +1,38 @@
"""SQLAlchemy engine / Session 工厂与 FastAPI 依赖。
注意:本文件属于 Infrastructure 层,是唯一允许出现 engine 创建的地方之一。
业务 Service / Domain 禁止直接使用 Session —— 一律通过 Repository。
"""
from __future__ import annotations
from collections.abc import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from app.core.config import get_settings
_settings = get_settings()
engine = create_engine(
_settings.database_url,
echo=_settings.sqlalchemy_echo,
future=True,
)
SessionLocal = sessionmaker(
bind=engine,
autoflush=False,
expire_on_commit=False,
class_=Session,
)
def get_session() -> Generator[Session, None, None]:
"""FastAPI 依赖:每个请求一个 Session,请求结束自动关闭。
供 Repository 实现内部使用;业务层通过 Repository 间接访问。
"""
with SessionLocal() as session:
yield session
+21
View File
@@ -0,0 +1,21 @@
"""FastAPI 应用入口。
启动:cd backend && uv run uvicorn app.main:app --reload
"""
from __future__ import annotations
from fastapi import FastAPI
from app.api.router import api_router
from app.core.config import get_settings
settings = get_settings()
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="A股个人量化研究平台 API(Qlib 引擎 / Tushare 数据源)",
)
app.include_router(api_router, prefix=settings.api_prefix)
+6
View File
@@ -0,0 +1,6 @@
"""量化研究层:研究服务与 Qlib 适配。
约束(AGENT.md §14/§15):
- 业务层禁止直接 import qlib;Qlib 相关代码只允许出现在 qlib_adapter/ 下
- 禁止修改 site-packages/qlib 源码,一律 Adapter / Wrapper / Extension
"""
@@ -0,0 +1,5 @@
"""Qlib 适配层(Phase 2 实现)。
预期模块:dataset.py / feature.py / model.py / backtest.py / provider.py。
业务代码形态应是 backtest_service.run(spec),而不是散落的 qlib.init(...)。
"""