stocklens/storage/ & stocklens/repositories/ — 数据库 & ORM

职责

SQLAlchemy ORM 层：数据库连接管理、18 张表定义、9 个 Repository 封装 CRUD。

文件清单

storage/

文件	职责
`engine.py`	`DatabaseManager` 单例 + PRAGMA 配置 + `save_scoring_snapshots()` + `_run_alembic_upgrade`（启动时自动跑 alembic, G2）
`models.py`	SQLAlchemy Model 定义（改这里必须配 Alembic migration，铁律 #3）
`helpers.py`	scoring 快照保存等辅助函数（含 `build_raw_result` / `safe_json_dumps`）
`analysis_snapshot.py`	2026-05 G10 新增 — 从 `utils/data_processing.py` 迁入：`extract_fundamental_context` / `extract_fundamental_detail_fields`（懂 `context_snapshot.enhanced_context.fundamental_context.earnings.data` 的嵌套 schema）
`__init__.py`	导出

stocklens/utils/data_processing.py 保留 normalize_model_used / parse_json_field 两个真正的纯工具函数，并把 extract_fundamental_context / extract_fundamental_detail_fields 作为 backward-compat re-export 暴露 —— 旧 import 路径不破坏。

retention/（G1 2026-05）

文件	职责
`policies.py`	`RetentionPolicy` dataclass + `DEFAULT_POLICIES` 列表（16 张表的保留窗口）
`service.py`	`RetentionService` — 运行 dry-run / execute；`RetentionReport` 结构化结果
`__init__.py`	公开门面

默认保留窗口：

表	窗口	理由
`news_intel`	90 天	时效性
`market_news_archive`	60 天	大盘新闻短时效
`llm_usage`	30 天明细	计费 / 审计；后续聚合 180 天（计划）
`analysis_history`	365 天	报告 1 年；后续保留 metadata-only（计划）
`chip_distribution_history`	180 天	A 股筹码
`tech_indicator_snapshot`	90 天	可重算
`social_sentiment_snapshot` / `institutional_snapshot` / `market_env_snapshot` / `correlation_snapshot` / `finviz_forward_snapshot` / `iv_snapshot` / `analyst_target_snapshot` / `shares_outstanding_snapshot`	180 天	各类快照
`fundamental_snapshot`	365 天	4 个季度
`backtest_results`	365 天	聚合到 `BacktestSummary` 永久保留

永久保留（不出现在 policy 表）：stock_daily / stock_weekly / stock_bars / stock_quote_snapshot / macro_snapshot / pe_history / portfolio_* / conversation_messages。

API 入口：

GET  /api/v1/system/retention            # dry-run，只统计
POST /api/v1/system/retention/execute    # 真删（破坏性）

编程入口：

from stocklens.storage.retention import RetentionService
from stocklens.storage.engine import get_db

svc = RetentionService(db_manager=get_db())
report = svc.run()              # dry-run
print(report.total_deletable)   # 多少行可以删
print(report.plans[0].to_dict())  # 每张表细节

report = svc.run(execute=True)  # 真删
print(report.total_deleted)

测试：tests/unit/test_retention.py（6 用例锁定 dry-run / execute / 缺表 skip / 单 policy 错误不阻塞 / JSON 序列化等 contract）。

G1 当前阶段：仅提供 dry-run + execute API，没有自动调度——运维需要手动调用或脚本化。后续会接入 APScheduler 跑日级清理 + Parquet 归档。

alembic/（项目根目录，G2 2026-05）

文件	职责
`alembic.ini`	Alembic 主配置；默认 DB URL `sqlite:///./data/stock_analysis.db`，env var `ALEMBIC_DATABASE_URL` 可覆盖
`alembic/env.py`	把 Alembic 接到 `Base.metadata`；优先复用 `DatabaseManager` 打开的 connection（避免 SQLite WAL 双 writer）；`render_as_batch=True` 让 SQLite ALTER 走 batch 模式
`alembic/versions/49a950e262eb_baseline_existing_schema.py`	Baseline 永不修改：`upgrade()` 调 `Base.metadata.create_all`，对新 DB 创建所有 30 张表，对已有 prod DB 是幂等的 no-op + stamp `alembic_version`
`alembic/versions/<future>_*.py`	后续 schema change 用 `alembic revision --autogenerate -m "..."` 生成

启动时 DatabaseManager.__init__ 自动 alembic upgrade head，失败 fall back create_all（loud warning）。详见 docs/dev_guide.md "DB Schema Migrations" 节。

repositories/

文件	行数	职责
`base.py`	~95	2026-05 新增（M3）：`BaseRepository` 抽出共享 session 管理逻辑（三段式 session_factory 解析 + `_session_scope` 上下文管理器）
`analysis_repo.py`	~200	分析结果 CRUD（继承 `BaseRepository`）
`stock_repo.py`	~600	日线/周线 + 多周期 K 线 (`stock_bars`) + 行情快照 (`stock_quote_snapshot`) CRUD
`news_repo.py`	~200	新闻情报 CRUD
`fundamental_repo.py`	~200	基本面数据 CRUD
`macro_repo.py`	~150	宏观数据 CRUD
`backtest_repo.py`	~200	回测结果 CRUD
`conversation_repo.py`	~120	对话历史 CRUD（继承 `BaseRepository`）
`llm_usage_repo.py`	~150	LLM 用量审计 CRUD（继承 `BaseRepository`）
`portfolio_repo.py`	~300	组合/交易/快照 CRUD
`snapshot_repo.py`	~470	2026-05 新增：7 张每日快照表统一 UPSERT (`SnapshotRepository`)

`BaseRepository`（base.py）— M3 共享 session 管理（2026-05）

之前 AnalysisRepository / LLMUsageRepository / ConversationRepository 各自重复 32 行的 session 管理样板代码（triple-fallback 解析 + _session_scope context manager）。BaseRepository 把它们抽出，子类只需 class FooRepository(BaseRepository): 即可继承 _get_session() / _session_scope() / _db_manager 字段，构造签名（session_factory=None, db_manager=None）也保留不变 — 现有调用点零改动。

class BaseRepository:
    def __init__(self, session_factory=None, db_manager=None):
        # triple-fallback: explicit factory > db_manager.get_session > DatabaseManager singleton
        ...

    def _get_session(self) -> Session: ...
    _session = _get_session  # H6 alias for repos that commit inline
    def _session_scope(self) -> Iterator[Session]: ...  # 委托 db_manager 或回退到 commit/rollback/close

`StubRepository`（base.py）— H6 P13 迁移占位基类（2026-05）

4 个 P13 占位 repo（FundamentalRepository / MacroRepository / NewsRepository / StockRepository）之前各自重复同一份 5 行的 no-op 构造函数 + 一份 module-level _shim_warn 函数。StubRepository 把它们收口：

class StubRepository:
    def __init__(self, session_factory=None, db_manager=None):
        # accepted for API parity with BaseRepository; never used
        self._session_factory = session_factory
        self._db_manager = db_manager

    def _shim_warn(self, method: str, code: str = "") -> None:
        logger.debug("[%s shim] %s(%s) -> no-op (P13 → tickbridge-server)",
                     type(self).__name__, method, code, ...)

存在意义：让「这是过渡期占位、不是真实 repo」在类型层面立即可见。P14 删除时一行 rg "StubRepository" 列出全部目标。

H6 完成的迁移

Repo	之前	现在
`AnalysisRepository`	32 行 ctor + scope (M3)	`class AnalysisRepository(BaseRepository):`
`LLMUsageRepository`	同上	`class LLMUsageRepository(BaseRepository):`
`ConversationRepository`	同上	`class ConversationRepository(BaseRepository):`
`ApiKeyRepository`	12 行 ctor + `_session()`	`class ApiKeyRepository(BaseRepository):`
`AuditLogRepository`	同上	`class AuditLogRepository(BaseRepository):`
`FundamentalRepository`	6 行 ctor + module `_shim_warn`	`class FundamentalRepository(StubRepository):`
`MacroRepository`	同上	`class MacroRepository(StubRepository):`
`NewsRepository`	同上	`class NewsRepository(StubRepository):`
`StockRepository`	6 行 ctor + module `_shim_warn` (override 保留)	`class StockRepository(StubRepository):`

5 个 BaseRepository 子类 + 4 个 StubRepository 子类 = 9 个 repo 全部用基类。剩余 3 个（backtest_repo / portfolio_repo / fx_rate_repo / snapshot_repo）使用 self.db = db_manager 模式（不同的 attribute 名），暂时保留各自的 ctor 形状。

回归测试 tests/unit/repositories/test_base_repository.py（19 用例）锁定基类行为：构造签名、session 解析三段式、scope commit/rollback/close、stub _shim_warn 输出格式 + 9 个 repo 的 isinstance 关系。

核心类

`DatabaseManager`（engine.py）

单例模式，管理 SQLite 连接和会话。

SQLite PRAGMA 配置（每次新连接自动应用）

_setup_sqlite_pragmas() 注册了 SQLAlchemy connect 事件监听器，在每个新物理连接上执行：

PRAGMA	值	作用
`journal_mode`	`WAL`	写者不阻塞读者，并发读 2-5×
`synchronous`	`NORMAL`	比 FULL 快 ~2×，分析场景下持久性足够
`busy_timeout`	`5000`	锁竞争时等待 5 秒而不是立即失败
`cache_size`	`-64000`	每连接 64MB 页缓存
`temp_store`	`MEMORY`	临时表/索引保留在内存中
`foreign_keys`	`ON`	启用 FK 约束（SQLite 默认关闭）
`mmap_size`	`268435456`	256MB 内存映射 IO，加速重读查询
`wal_autocheckpoint`	`2000`	每 ~8 MiB 触发 WAL → 主库 checkpoint（SQLite 默认 1000 页 / 4 MiB）。F4：批量 upsert pattern 下默认值 ~每 0.5 只股票就 checkpoint 一次，造成 spurious 写延迟。
`optimize`	—	每次连接打开时跑一次（cheap，无变化时 SQLite 跳过）；`_cleanup_engine` atexit 时再跑一次配合 `wal_checkpoint(TRUNCATE)`。

_cleanup_engine 关闭流程（F4）：除了 engine.dispose()，atexit 时还跑 PRAGMA optimize + PRAGMA wal_checkpoint(TRUNCATE)，刷新 planner 统计并把 WAL 截到零长度——避免重启后留下 multi-MB 的 WAL 反复重放。两步都是 best-effort（失败仅 debug log，不抛）。

关键方法

DatabaseManager.initialize(db_path: str = "data/stock_analysis.db") -> None — 初始化数据库，创建所有表
DatabaseManager.get_session() -> Session — 获取 SQLAlchemy Session
DatabaseManager.get_engine() -> Engine — 获取 Engine 实例

批量 Upsert

save_dataframe（日线）/ save_weekly_data（周线）/ save_pe_history（PE 历史）使用 SQLite INSERT ... ON CONFLICT DO UPDATE 一次性 upsert，相比逐行 select-then-insert 快 10-20×。

使用模式

DatabaseManager.initialize()
with DatabaseManager.get_session() as session:
    repo = AnalysisRepo(session)
    result = repo.get_latest("AAPL")

ORM Models（models.py）

表清单（20 张）

表清单分两组：核心业务表 + K 线/行情数据表（详见下表）。

K 线 / 行情数据表（4 张，US-only 多周期支持）

Model 类	表名	主键/唯一约束	关键字段	范围
`StockDaily`	`stock_daily`	(code, date)	OHLCV + amount + pct_chg + ma5/10/20 + volume_ratio	日线（所有市场）
`StockWeekly`	`stock_weekly`	(code, date)	OHLC + volume	周线（resample 自日线）
`StockBars`	`stock_bars`	(code, interval, datetime)	OHLCV + amount + pct_chg + ma5/10/20 + interval	60m / 1mo / 1q / 1y 多周期
`StockQuoteSnapshot`	`stock_quote_snapshot`	(code, snapshot_date)	35 字段：价格/估值/股本/股息/交易元数据	每日行情面板（雪球 quote.json 全字段持久化）

stock_bars 设计理念：单表 + interval 字段，扩展新周期（如未来加 5m/120m）只需追加 _ALLOWED_INTERVALS，不需要新表 / migration。当前 US-only 启用 60m/1mo/1q/1y 四个周期，1d/1w 保留为 alias（读 stock_daily/stock_weekly）。

stock_quote_snapshot 用途：每日跑批落盘"基本数据面板"，离线场景可直接从 SQLite 回放 PE/股息率/Beta 等历史，无需重新触达上游 API。Idempotent (一天一行/股，重跑覆盖)。

每日快照表（7 张，2026-05 新增）

Model 类	表名	唯一约束	关键字段	用途
`SocialSentimentSnapshot`	`social_sentiment_snapshot`	(code, snapshot_date)	reddit_score / reddit_mentions / x_mentions / polymarket_probability + raw_payload	Reddit/X/Polymarket 情绪曲线
`InstitutionalSnapshot`	`institutional_snapshot`	(code, snapshot_date, snapshot_type)	analyst_target_price / inst_holders_count / insider_net_shares + payload(JSON)	OpenBB 13F / insider / analyst 拆三类持久化
`ChipDistributionHistory`	`chip_distribution_history`	(code, snapshot_date)	profit_ratio / avg_cost / concentration_70 / concentration_90	A 股筹码分布历史曲线
`TechIndicatorSnapshot`	`tech_indicator_snapshot`	(code, snapshot_date)	MA5-200 / RSI / MACD / Bollinger / ATR / Sharpe / Beta / OBV / VWAP / Fibonacci JSON	历史技术指标快照（避免回测时重算）
`MarketNewsArchive`	`market_news_archive`	(source, content_hash)	source(jin10/eastmoney) / headline / content / published_at	大盘新闻存档（按 md5 去重）
`MarketEnvSnapshot`	`market_env_snapshot`	(snapshot_date, region)	sp500_pct / nasdaq_pct / vix / pe_percentile / regime + payload(JSON)	市场环境历史曲线（每天 1 行/region）
`CorrelationSnapshot`	`correlation_snapshot`	(code, snapshot_date)	spy_corr_20d/60d / qqq_corr_20d/60d	相关性漂移监测

写入入口：stocklens/pipeline/stages/persistence_stage.py 的 persist_all_snapshots(...)，在 _collect_full_context 里跑完 enrichment 之后、生成元数据报告之前统一调用。失败 → WARNING 不阻断主流程。Idempotent：所有表用 INSERT ... ON CONFLICT DO UPDATE，重跑当日覆盖。

查询接口（SnapshotRepository）：每张表都提供 save_* + get_recent_*(code, days=N) 一对方法，统一返回 ORM 对象列表。

`unit_of_work()` — Session 复用上下文（F3，2026-05）

with snapshot_repo.unit_of_work():
    snapshot_repo.save_chip_distribution(...)
    snapshot_repo.save_tech_indicators(...)
    snapshot_repo.save_correlation(...)
    # ...一只股票典型 6-9 次 save_*

做什么：让一只股票的所有 save_* 共享同一个 Session 对象，避免每次 save 都重新 acquire 一个 connection-pool slot + 跑一遍 PRAGMA busy_timeout/synchronous/... 设置。

不做什么：不是 all-or-nothing 的 atomic batch。每个 save_* 内部仍然独立 commit()。原因：SQLite 的 SAVEPOINT 语义与 PG/MySQL 不同——RELEASE SAVEPOINT 后再外层 ROLLBACK 不会撤销 SAVEPOINT 内的写入；如果走 SAVEPOINT 的话一行 IntegrityError 就会污染整批。所以选择保留 per-row commit 来换 per-row 的 best-effort resilience。

真要原子：用 DatabaseManager.session_scope() 直接管理事务（牺牲 per-row 容错）。

线程安全：每个线程通过 threading.local 持有独立 bound session；嵌套调用 unit_of_work() 复用外层 session（idempotent re-entry）。

回归测试 tests/unit/repositories/test_snapshot_repo.py::TestUnitOfWork（6 用例）锁定语义。

核心业务表（其余 16 张）

Model 类	表名	主键	关键字段
`StockInfo`	`stock_info`	id	code, name, market, sector, industry
`AnalysisResult`	`analysis_results`	id	code, date, score, prediction, advice, report_json, model_used
`NewsIntel`	`news_intel`	id	code, title, content, source, pub_date, dimension
`FundamentalData`	`fundamental_data`	id	code, period, revenue, net_income, roe, gross_margin
`MacroData`	`macro_data`	id	indicator, value, date, source
`BacktestResult`	`backtest_results`	id	code, analysis_date, eval_date, direction_correct, return_pct
`Conversation`	`conversations`	id	session_id, role, content, model, created_at
`LLMUsage`	`llm_usage`	id	model, tokens_in, tokens_out, cost, latency_ms, created_at
`PortfolioAccount`	`portfolio_accounts`	id	name, currency, description, created_at
`PortfolioTransaction`	`portfolio_transactions`	id	account_id, code, action, shares, price, fee, date
`PortfolioSnapshot`	`portfolio_snapshots`	id	account_id, date, total_value, cash, positions_json
`SystemConfig`	`system_config`	id	key, value, category, updated_at
`TaskQueue`	`task_queue`	id	task_id, task_type, status, params_json, result_json, created_at
`UserSession`	`user_sessions`	id	session_token, user_id, expires_at, created_at
`RateLimit`	`rate_limits`	id	ip, endpoint, count, window_start
`StockSearchCache`	`stock_search_cache`	id	query, results_json, cached_at

Repository 类

所有 Repository 遵循统一模式：

class XxxRepo:
    def __init__(self, session: Session): ...
    def save(self, data: dict) -> Model: ...
    def get_by_code(self, code: str, ...) -> Optional[Model]: ...
    def get_latest(self, code: str) -> Optional[Model]: ...
    def query(self, filters: dict) -> List[Model]: ...
    def delete(self, id: int) -> bool: ...

`AnalysisRepo`（analysis_repo.py）

save_result(code: str, date: str, result: dict) -> AnalysisResult — 保存分析结果
get_latest(code: str) -> Optional[AnalysisResult] — 获取最新分析
get_by_date(code: str, date: str) -> Optional[AnalysisResult] — 按日期查询
get_history(code: str, limit: int = 10) -> List[AnalysisResult] — 历史记录
get_all_by_date(date: str) -> List[AnalysisResult] — 某日所有分析
count_by_date(date: str) -> int — 某日分析数量

`StockRepository`（stock_repo.py）

日线 / 周线（与 stock_daily / stock_weekly 表配对）：

save_dataframe(df, code, data_source) -> int — 批量 upsert 日线（SQLite ON CONFLICT），返回新增条数
save_weekly_data(df, code, data_source) -> int — 批量 upsert 周线
get_range(code, start, end) -> List[StockDaily] — 按日期范围读日线
get_weekly_range(code, start, end) -> List[StockWeekly] — 周线
get_analysis_context(code) -> Optional[dict] — 分析所用上下文（today/yesterday）
get_latest_date(code) -> Optional[date] — 最新日期

多周期 K 线（stock_bars 表）：

save_bars(df, code, interval, data_source) -> int — 批量 upsert 任意周期，interval ∈ {'60m','1mo','1q','1y'}（也接受 1h/hour/monthly/quarterly/yearly 等别名）
get_bars_range(code, interval, start_dt, end_dt) -> List[StockBars] — 按时间范围读
get_latest_bar_datetime(code, interval) -> Optional[datetime] — 增量同步用
get_bars_count(code, interval) -> int — 统计

行情快照（stock_quote_snapshot 表）：

save_quote_snapshot(code, quote, snapshot_date=None, data_source='Xueqiu') -> bool — 写入今日行情面板（idempotent，重跑同日覆盖）
get_quote_snapshot_range(code, start_date, end_date) -> List[StockQuoteSnapshot] — 历史快照
get_latest_quote_snapshot(code) -> Optional[StockQuoteSnapshot] — 最新一条

Allowed intervals 常量：stock_repo._ALLOWED_INTERVALS = {'60m','1mo','1q','1y','1d','1w'}，新增周期只需追加该集合（无 DDL）。

`NewsRepository`（news_repo.py）

save_news_intel(code, name, dimension, query, response, query_context) -> int — 保存搜索情报（按 url 去重）
get_recent_news(code, days, limit) -> List[NewsIntel]

`FundamentalRepository`（fundamental_repo.py）

save_snapshot(query_id, code, payload, source_chain, coverage) — 写入基本面快照
save_pe_history(code, pe_data) -> int — 批量 upsert PE 历史（SQLite ON CONFLICT）
get_pe_history(code, days) — 读取 PE 历史
save_finviz_forward(code, data) -> List[changes]
get_finviz_forward(code) -> Optional[FinvizForwardSnapshot]

`MacroRepo`（macro_repo.py）

save_macro(indicator: str, value: float, date: str) -> MacroData — 保存宏观指标
get_latest(indicator: str) -> Optional[MacroData] — 最新值

`BacktestRepo`（backtest_repo.py）

save_result(result: dict) -> BacktestResult — 保存回测结果
get_by_code(code: str, limit: int = 20) -> List[BacktestResult] — 按股票查询
get_pending(code: str = None) -> List[AnalysisResult] — 获取待回测的分析

`ConversationRepo`（conversation_repo.py）

save_message(session_id: str, role: str, content: str, model: str = None) -> Conversation
get_history(session_id: str, limit: int = 50) -> List[Conversation]
delete_session(session_id: str) -> int — 删除会话，返回删除条数
list_sessions(limit: int = 20) -> List[dict] — 会话列表

`LLMUsageRepo`（llm_usage_repo.py）

log_usage(model: str, tokens_in: int, tokens_out: int, cost: float, latency_ms: int) -> LLMUsage
get_usage_stats(days: int = 30) -> dict — 用量统计

`PortfolioRepo`（portfolio_repo.py）

create_account(name: str, currency: str = "USD") -> PortfolioAccount
add_transaction(account_id: int, code: str, action: str, shares: float, price: float) -> PortfolioTransaction
get_transactions(account_id: int) -> List[PortfolioTransaction]
save_snapshot(account_id: int, snapshot: dict) -> PortfolioSnapshot
get_snapshots(account_id: int, days: int = 30) -> List[PortfolioSnapshot]

依赖关系

依赖：sqlalchemy、stocklens/config/
被依赖：pipeline、analyzer、services、api、notification

配置项

变量	默认值	说明
数据库路径	`data/stock_analysis.db`	硬编码在 `DatabaseManager.initialize()`；StockLens 独占，tickbridge-server 使用独立的 `data/tickbridge.db`

注意事项

DatabaseManager 是单例，必须在使用任何 Repository 前调用 initialize()
所有 Repository 需要传入 Session，推荐使用 with DatabaseManager.get_session() as session: 上下文管理器
StockRepo.save_daily_data() 自动去重（按 code + date），已存在的记录会跳过
NewsRepo.save_news() 同样自动去重（按 code + title + pub_date）
SQLite 不支持并发写入，但已启用 WAL 模式，写不阻塞读
批量场景请使用 upsert 方法（save_dataframe / save_weekly_data / save_pe_history），而不是循环 add()
任何新表都建议加 UniqueConstraint 配合 INSERT ... ON CONFLICT DO UPDATE 模式

性能优化（2026-05）

DataFrame 批量写入向量化：stock_repo.{save_dataframe, save_weekly_data, save_bars} 改用 _normalize_date_column + _df_to_records_with_meta 替换 for _, row in df.iterrows(): 循环；pd.to_datetime(..., errors='coerce') 一次性把异构日期列规范化为 date 对象，随后 to_dict('records') 一次性产出 list[dict]。pandas 官方文档建议的最快路径，10-50× 提速。
news_repo.save_news_intel 改 UPSERT：原先是 for item in results: 内做 select-then-insert + savepoint，30 stocks × 30 items ≈ 1800 round-trips。改为单条 INSERT ... ON CONFLICT(url) DO UPDATE + 本地 url 去重 + case 表达式实现"非空覆盖、空值保留" 语义。返回值语义改为 "rows processed"（不再区分新增 vs 更新）。
analysis_repo.count_by_code 改 SQL COUNT()：原代码 get_history(limit=1000) 把含 raw_result / news_content / context_snapshot 大字段的 ORM 对象全部加载只为了 len()。改为 select(func.count(AnalysisHistory.id)) + where created_at >= cutoff。
新增复合索引（保留旧的单列索引）：
- MacroSnapshot: Index('ix_macro_indicator_date', 'indicator', 'data_date') — 优化 "indicator=X AND data_date BETWEEN ..." 范围扫描。
- LLMUsage: Index('ix_llm_usage_model_time', 'model', 'called_at') 与 Index('ix_llm_usage_code_time', 'stock_code', 'called_at') — 匹配 token-by-model-since-X 与 cost-per-stock 等典型审计查询。

缓存层（utils/cache.py）

二级缓存（实现位于 kvcache/，业务通过 stocklens.utils.cache_bootstrap 注册 namespace）：

L1 — 内存 (kvcache.MemoryBackend)：进程内内存缓存，每个 namespace 自带 TTL + 容量上限 + 命中率统计
L2 — SQLite (kvcache.SqliteBackend)：跨进程文件缓存（data/cache.db，PRAGMA WAL），多个 worker 共享
TieredBackend：L1 + L2 复合，读 L1 → L2 → miss；写时同时写入

监控

调用 from kvcache import collect_overview, log_cache_health 可获取全局画像：

overview = collect_overview()
# 返回：
# {
#   "manager": {"namespaces": N, "total_hits": ..., "hit_rate": 0.957, ...},
#   "namespaces": {"bollinger": {...}, "spy_qqq": {...}, ...},
#   "backends": [{"name": "default_tiered", "l1": {...}, "l2": {...}}],
# }

log_cache_health()  # 直接写入 INFO 日志，适合放在长任务起止前后

适合在批量分析前后调用，留下命中率与容量轨迹。

注意事项

缓存失败一律 fail-open：任何 L1/L2 异常被吞为 warning，不阻断主流程
L2 SQLite 通过 get_shared_sqlite_cache() 单例模式共享同一文件
start_cache_janitor(interval=3600) 启动后台清理线程，定期 prune 过期项
长任务建议监控 L1 total_size 与 L2 total_entries，单调递增意味着可能漏 prune
2026-05 修复 engine.py 静态注解缺失：session_scope() -> "Generator[Session, None, None]" 中的 Generator 未 import；ruff F821 暴露后通过 if TYPE_CHECKING: from typing import Generator 修补（Session 已 runtime import 不重复）。

FilesExpand file tree

storage.md

Latest commit

History

storage.md

File metadata and controls

stocklens/storage/ & stocklens/repositories/ — 数据库 & ORM

职责

文件清单

storage/

retention/（G1 2026-05）

alembic/（项目根目录，G2 2026-05）

repositories/

BaseRepository（base.py）— M3 共享 session 管理（2026-05）

StubRepository（base.py）— H6 P13 迁移占位基类（2026-05）

H6 完成的迁移

核心类

DatabaseManager（engine.py）

SQLite PRAGMA 配置（每次新连接自动应用）

关键方法

批量 Upsert

使用模式

ORM Models（models.py）

表清单（20 张）

K 线 / 行情数据表（4 张，US-only 多周期支持）

每日快照表（7 张，2026-05 新增）

unit_of_work() — Session 复用上下文（F3，2026-05）

核心业务表（其余 16 张）

Repository 类

AnalysisRepo（analysis_repo.py）

StockRepository（stock_repo.py）

NewsRepository（news_repo.py）

FundamentalRepository（fundamental_repo.py）

MacroRepo（macro_repo.py）

BacktestRepo（backtest_repo.py）

ConversationRepo（conversation_repo.py）

LLMUsageRepo（llm_usage_repo.py）

PortfolioRepo（portfolio_repo.py）

依赖关系

配置项

注意事项

性能优化（2026-05）

缓存层（utils/cache.py）

监控

注意事项

`BaseRepository`（base.py）— M3 共享 session 管理（2026-05）

`StubRepository`（base.py）— H6 P13 迁移占位基类（2026-05）

`DatabaseManager`（engine.py）

`unit_of_work()` — Session 复用上下文（F3，2026-05）

`AnalysisRepo`（analysis_repo.py）

`StockRepository`（stock_repo.py）

`NewsRepository`（news_repo.py）

`FundamentalRepository`（fundamental_repo.py）

`MacroRepo`（macro_repo.py）

`BacktestRepo`（backtest_repo.py）

`ConversationRepo`（conversation_repo.py）

`LLMUsageRepo`（llm_usage_repo.py）

`PortfolioRepo`（portfolio_repo.py）