diff --git a/app/services/cache.py b/app/services/cache.py index 2f62d20..93e73dd 100644 --- a/app/services/cache.py +++ b/app/services/cache.py @@ -10,6 +10,7 @@ from sqlalchemy.orm import Session from app.models import MarketData, ScheduledTask, SymbolTimestamp from app.config import CACHE_TTL_SECONDS +from app.storage_manager import get_storage_manager logger = logging.getLogger(__name__) @@ -65,11 +66,12 @@ def check_cache_status( } -def save_market_data(db: Session, symbol: str, data: Dict) -> MarketData: +def _save_to_sqlite(db: Session, symbol: str, data: Dict) -> MarketData: """ - 保存采集结果到缓存，并同步更新合约时间戳。 + 将采集结果写入 SQLite 缓存，并同步更新合约时间戳。 Args: + db: 数据库会话 symbol: 品种代码 data: 采集脚本返回的完整数据 @@ -106,7 +108,7 @@ def save_market_data(db: Session, symbol: str, data: Dict) -> MarketData: db.add(record) # 更新合约时间戳 - update_symbol_timestamp(db, symbol, data.get("type", "futures"), now) + _update_symbol_timestamp_sqlite(db, symbol, data.get("type", "futures"), now) db.commit() logger.info(f"缓存已更新: {symbol}, {len(data.get('timeframes', {}))} 个周期") @@ -118,8 +120,8 @@ def save_market_data(db: Session, symbol: str, data: Dict) -> MarketData: ).order_by(MarketData.fetched_at.desc()).first() -def update_symbol_timestamp(db: Session, symbol: str, data_type: str, refresh_time: datetime) -> None: - """更新或创建合约时间戳记录""" +def _update_symbol_timestamp_sqlite(db: Session, symbol: str, data_type: str, refresh_time: datetime) -> None: + """更新或创建合约时间戳记录（SQLite 兜底）。""" timestamp_record = db.query(SymbolTimestamp).filter_by( symbol=symbol, data_type=data_type @@ -136,12 +138,10 @@ def update_symbol_timestamp(db: Session, symbol: str, data_type: str, refresh_ti refresh_count=1 ) db.add(timestamp_record) - - db.commit() -def get_symbol_timestamp(db: Session, symbol: str, data_type: str = "futures") -> Optional[datetime]: - """获取合约最后刷新时间""" +def _get_symbol_timestamp_sqlite(db: Session, symbol: str, data_type: str = "futures") -> Optional[datetime]: + """获取合约最后刷新时间（SQLite 兜底）。""" record = db.query(SymbolTimestamp).filter_by( symbol=symbol, data_type=data_type @@ -149,41 +149,7 @@ def get_symbol_timestamp(db: Session, symbol: str, data_type: str = "futures") - return record.last_refresh_at if record else None -def needs_refresh(db: Session, symbol: str, data_type: str = "futures", threshold_seconds: int = 300) -> bool: - """ - 检查合约是否需要刷新（数据是否超过阈值时间） - - Args: - db: 数据库会话 - symbol: 品种代码 - data_type: 数据类型 - threshold_seconds: 阈值时间（秒），默认300秒（5分钟） - - Returns: - True 表示需要刷新，False 表示数据仍然新鲜 - """ - last_refresh = get_symbol_timestamp(db, symbol, data_type) - if last_refresh is None: - return True # 从未刷新过，需要刷新 - - age = (datetime.now() - last_refresh).total_seconds() - return age > threshold_seconds - - -def get_latest_cached( - db: Session, - symbol: str, - data_type: str = "futures", - period: Optional[str] = None, -) -> List[MarketData]: - """获取最新缓存数据""" - query = db.query(MarketData).filter_by(symbol=symbol, data_type=data_type) - if period: - query = query.filter_by(period=period) - return query.order_by(MarketData.fetched_at.desc()).all() - - -def get_cached_data( +def _get_from_sqlite( db: Session, symbol: str, data_type: str = "futures", @@ -192,7 +158,7 @@ def get_cached_data( max_candles: int = 100, ) -> Optional[Dict]: """ - 从缓存中获取完整的多周期数据。 + 从 SQLite 缓存中获取完整的多周期数据。 Args: db: 数据库会话 @@ -277,6 +243,185 @@ def get_cached_data( } +def save_market_data(db: Session, symbol: str, data: Dict) -> Optional[MarketData]: + """ + 保存采集结果到缓存，优先写入 StorageManager（Redis/MySQL），失败时降级到 SQLite。 + + Args: + symbol: 品种代码 + data: 采集脚本返回的完整数据 + + Returns: + SQLite 写入时返回 MarketData 记录；StorageManager 成功时返回 None + """ + storage = get_storage_manager() + data_type = data.get("type", "futures") + now = datetime.now() + + if storage.is_available(): + try: + if storage.save_market_data(symbol, data): + storage.save_symbol_timestamp(symbol, data_type, now) + return None + else: + raise Exception("save_market_data returned False") + except Exception as e: + logger.warning(f"StorageManager 写入失败，降级到 SQLite: {e}") + + return _save_to_sqlite(db, symbol, data) + + +def update_symbol_timestamp(db: Session, symbol: str, data_type: str, refresh_time: datetime) -> None: + """更新或创建合约时间戳记录，优先使用 StorageManager。""" + storage = get_storage_manager() + + if storage.is_available(): + try: + storage.save_symbol_timestamp(symbol, data_type, refresh_time) + return + except Exception as e: + logger.warning(f"StorageManager 时间戳写入失败，降级到 SQLite: {e}") + + _update_symbol_timestamp_sqlite(db, symbol, data_type, refresh_time) + db.commit() + + +def get_symbol_timestamp(db: Session, symbol: str, data_type: str = "futures") -> Optional[datetime]: + """获取合约最后刷新时间，优先使用 StorageManager。""" + storage = get_storage_manager() + + if storage.is_available(): + try: + result = storage.get_symbol_timestamp(symbol, data_type) + if result: + return datetime.fromisoformat(result["last_refresh_at"]) + except Exception as e: + logger.warning(f"StorageManager 时间戳读取失败，降级到 SQLite: {e}") + + return _get_symbol_timestamp_sqlite(db, symbol, data_type) + + +def needs_refresh(db: Session, symbol: str, data_type: str = "futures", threshold_seconds: int = 300) -> bool: + """ + 检查合约是否需要刷新（数据是否超过阈值时间） + + Args: + db: 数据库会话 + symbol: 品种代码 + data_type: 数据类型 + threshold_seconds: 阈值时间（秒），默认300秒（5分钟） + + Returns: + True 表示需要刷新，False 表示数据仍然新鲜 + """ + last_refresh = get_symbol_timestamp(db, symbol, data_type) + if last_refresh is None: + return True # 从未刷新过，需要刷新 + + age = (datetime.now() - last_refresh).total_seconds() + return age > threshold_seconds + + +def get_latest_cached( + db: Session, + symbol: str, + data_type: str = "futures", + period: Optional[str] = None, +) -> List[MarketData]: + """获取最新缓存数据""" + query = db.query(MarketData).filter_by(symbol=symbol, data_type=data_type) + if period: + query = query.filter_by(period=period) + return query.order_by(MarketData.fetched_at.desc()).all() + + +def _merge_storage_market_data( + symbol: str, + data_type: str, + periods: Optional[List[str]], + storage_results: Dict[str, Optional[Dict]], +) -> Optional[Dict]: + """将 StorageManager 按周期返回的数据合并为 get_cached_data 标准格式。""" + timeframes = {} + current_price = None + newest = None + + for period, result in storage_results.items(): + if not result: + continue + candles = result.get("candles", []) + timeframes[period] = candles + if current_price is None: + current_price = result.get("current_price") + timestamp_raw = result.get("timestamp") + if timestamp_raw: + try: + timestamp_dt = datetime.fromisoformat(timestamp_raw) + if newest is None or timestamp_dt > newest: + newest = timestamp_dt + except Exception: + pass + + if not timeframes: + return None + + if newest is None: + newest = datetime.now() + + now = datetime.now() + is_fresh = (now - newest).total_seconds() < CACHE_TTL_SECONDS + + return { + "symbol": symbol, + "type": data_type, + "current_price": current_price, + "timestamp": newest.isoformat(), + "timeframes": timeframes, + "is_fresh": is_fresh, + "fetched_at": newest.isoformat(), + } + + +def get_cached_data( + db: Session, + symbol: str, + data_type: str = "futures", + periods: Optional[List[str]] = None, + end_time: Optional[datetime] = None, + max_candles: int = 100, +) -> Optional[Dict]: + """ + 从缓存中获取完整的多周期数据，优先使用 StorageManager，失败时降级到 SQLite。 + + Args: + db: 数据库会话 + symbol: 品种代码 + data_type: 数据类型 + periods: 周期列表 + end_time: 结束时间(可选),默认为当前时间 + max_candles: 每个周期最大K线数量,默认100 + + Returns: + 与采集脚本相同格式的数据，或 None + """ + storage = get_storage_manager() + + if storage.is_available(): + try: + storage_results = {} + query_periods = periods if periods else [] + for period in query_periods: + storage_results[period] = storage.get_market_data(symbol, data_type, period) + + merged = _merge_storage_market_data(symbol, data_type, periods, storage_results) + if merged: + return merged + except Exception as e: + logger.warning(f"StorageManager 读取失败，降级到 SQLite: {e}") + + return _get_from_sqlite(db, symbol, data_type, periods, end_time, max_candles) + + # ===== 定时任务管理 ===== def create_task( diff --git a/tests/test_cache_storage_integration.py b/tests/test_cache_storage_integration.py new file mode 100644 index 0000000..68a9962 --- /dev/null +++ b/tests/test_cache_storage_integration.py @@ -0,0 +1,286 @@ +""" +cache.py StorageManager 集成与 SQLite 降级逻辑测试 +""" +import json +import logging +from datetime import datetime +from unittest.mock import MagicMock, patch + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from app.models import Base, MarketData, SymbolTimestamp + + +@pytest.fixture +def db(): + """提供已创建表结构的内存 SQLite 会话。""" + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(bind=engine) + Session = sessionmaker(bind=engine) + session = Session() + yield session + session.close() + + +@pytest.fixture +def sample_data(): + """构造与采集脚本格式一致的测试数据。""" + return { + "type": "futures", + "current_price": 123.45, + "timestamp": datetime(2026, 7, 4, 10, 0, 0).isoformat(), + "timeframes": { + "5min": [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}], + "15min": [{"datetime": "2026-07-04T09:45:00", "open": 105, "close": 115}], + }, + } + + +@pytest.fixture +def storage_manager_mock(): + """构造默认可用的 StorageManager mock。""" + manager = MagicMock() + manager.is_available.return_value = True + manager.save_market_data.return_value = True + manager.save_symbol_timestamp.return_value = True + manager.get_market_data.return_value = None + manager.get_symbol_timestamp.return_value = None + return manager + + +def _seed_sqlite_market_data(db, symbol="AG2606", data_type="futures"): + """在 SQLite 中写入行情数据。""" + fetched_at = datetime(2026, 7, 4, 10, 0, 0) + db.add_all( + [ + MarketData( + symbol=symbol, + data_type=data_type, + period="5min", + candles_json=json.dumps( + [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}] + ), + current_price=123.45, + fetched_at=fetched_at, + candle_count=1, + ), + MarketData( + symbol=symbol, + data_type=data_type, + period="15min", + candles_json=json.dumps( + [{"datetime": "2026-07-04T09:45:00", "open": 105, "close": 115}] + ), + current_price=123.45, + fetched_at=fetched_at, + candle_count=1, + ), + ] + ) + db.commit() + + +def _seed_sqlite_symbol_timestamp(db, symbol="AG2606", data_type="futures"): + """在 SQLite 中写入合约时间戳。""" + db.add( + SymbolTimestamp( + symbol=symbol, + data_type=data_type, + last_refresh_at=datetime(2026, 7, 4, 10, 0, 0), + refresh_count=3, + ) + ) + db.commit() + + +class TestSaveMarketData: + """save_market_data 双写与降级测试。""" + + def test_uses_storage_manager_when_available(self, db, sample_data, storage_manager_mock): + """StorageManager 可用时优先写入存储层，并同步更新时间戳。""" + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + result = cache.save_market_data(db, "AG2606", sample_data) + + storage_manager_mock.is_available.assert_called_once() + storage_manager_mock.save_market_data.assert_called_once_with("AG2606", sample_data) + storage_manager_mock.save_symbol_timestamp.assert_called_once() + assert result is None + + sqlite_records = db.query(MarketData).filter_by(symbol="AG2606").all() + assert len(sqlite_records) == 0 + + def test_falls_back_to_sqlite_when_storage_unavailable(self, db, sample_data, storage_manager_mock): + """StorageManager 不可用时降级到 SQLite 写入。""" + storage_manager_mock.is_available.return_value = False + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + result = cache.save_market_data(db, "AG2606", sample_data) + + storage_manager_mock.save_market_data.assert_not_called() + storage_manager_mock.save_symbol_timestamp.assert_not_called() + + assert isinstance(result, MarketData) + sqlite_records = db.query(MarketData).filter_by(symbol="AG2606").all() + assert len(sqlite_records) == 2 + periods = {r.period for r in sqlite_records} + assert periods == {"5min", "15min"} + + def test_falls_back_to_sqlite_when_storage_save_fails(self, db, sample_data, storage_manager_mock): + """StorageManager 写入失败时降级到 SQLite 写入。""" + storage_manager_mock.save_market_data.return_value = False + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + result = cache.save_market_data(db, "AG2606", sample_data) + + storage_manager_mock.save_market_data.assert_called_once() + storage_manager_mock.save_symbol_timestamp.assert_not_called() + assert isinstance(result, MarketData) + assert db.query(MarketData).filter_by(symbol="AG2606").count() == 2 + + def test_falls_back_to_sqlite_when_storage_raises(self, db, sample_data, storage_manager_mock, caplog): + """StorageManager 写入异常时降级到 SQLite 写入并记录警告。""" + storage_manager_mock.save_market_data.side_effect = Exception("storage down") + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + with caplog.at_level(logging.WARNING, logger="app.services.cache"): + result = cache.save_market_data(db, "AG2606", sample_data) + + assert isinstance(result, MarketData) + assert db.query(MarketData).filter_by(symbol="AG2606").count() == 2 + assert "StorageManager 写入失败" in caplog.text + + +class TestGetCachedData: + """get_cached_data 读取与降级测试。""" + + def test_uses_storage_manager_and_merges_periods(self, db, storage_manager_mock): + """StorageManager 返回各周期数据时，合并为统一格式。""" + fetched_at = datetime.now() + + def mock_get_market_data(symbol, data_type, period): + return { + "current_price": 123.45, + "timestamp": fetched_at.isoformat(), + "candles": [{"datetime": f"2026-07-04T09:{50 if period == '5min' else 45}:00", "open": 100, "close": 110}], + } + + storage_manager_mock.get_market_data.side_effect = mock_get_market_data + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + result = cache.get_cached_data(db, "AG2606", "futures", ["5min", "15min"]) + + assert storage_manager_mock.get_market_data.call_count == 2 + storage_manager_mock.get_market_data.assert_any_call("AG2606", "futures", "5min") + storage_manager_mock.get_market_data.assert_any_call("AG2606", "futures", "15min") + + assert result is not None + assert result["symbol"] == "AG2606" + assert result["type"] == "futures" + assert result["current_price"] == 123.45 + assert result["timestamp"] == fetched_at.isoformat() + assert result["fetched_at"] == fetched_at.isoformat() + assert "5min" in result["timeframes"] + assert "15min" in result["timeframes"] + assert len(result["timeframes"]["5min"]) == 1 + assert result["is_fresh"] is True + + def test_falls_back_to_sqlite_when_storage_returns_none(self, db, storage_manager_mock): + """StorageManager 未命中时降级到 SQLite 读取。""" + storage_manager_mock.get_market_data.return_value = None + _seed_sqlite_market_data(db) + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + result = cache.get_cached_data(db, "AG2606", "futures", ["5min", "15min"]) + + assert result is not None + assert result["symbol"] == "AG2606" + assert "5min" in result["timeframes"] + assert "15min" in result["timeframes"] + + def test_falls_back_to_sqlite_when_storage_raises(self, db, storage_manager_mock, caplog): + """StorageManager 读取异常时降级到 SQLite 读取并记录警告。""" + storage_manager_mock.get_market_data.side_effect = Exception("storage down") + _seed_sqlite_market_data(db) + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + with caplog.at_level(logging.WARNING, logger="app.services.cache"): + result = cache.get_cached_data(db, "AG2606", "futures", ["5min"]) + + assert result is not None + assert "StorageManager 读取失败" in caplog.text + + +class TestSymbolTimestamp: + """合约时间戳函数封装测试。""" + + def test_update_symbol_timestamp_uses_storage_when_available(self, db, storage_manager_mock): + """StorageManager 可用时优先写入合约时间戳。""" + refresh_time = datetime(2026, 7, 4, 10, 0, 0) + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + cache.update_symbol_timestamp(db, "AG2606", "futures", refresh_time) + + storage_manager_mock.save_symbol_timestamp.assert_called_once_with("AG2606", "futures", refresh_time) + sqlite_record = db.query(SymbolTimestamp).filter_by(symbol="AG2606").first() + assert sqlite_record is None + + def test_update_symbol_timestamp_falls_back_to_sqlite(self, db, storage_manager_mock): + """StorageManager 不可用时降级到 SQLite 写入合约时间戳。""" + storage_manager_mock.is_available.return_value = False + refresh_time = datetime(2026, 7, 4, 10, 0, 0) + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + cache.update_symbol_timestamp(db, "AG2606", "futures", refresh_time) + + storage_manager_mock.save_symbol_timestamp.assert_not_called() + sqlite_record = db.query(SymbolTimestamp).filter_by(symbol="AG2606").first() + assert sqlite_record is not None + assert sqlite_record.last_refresh_at == refresh_time + assert sqlite_record.refresh_count == 1 + + def test_get_symbol_timestamp_uses_storage_when_available(self, db, storage_manager_mock): + """StorageManager 可用时优先读取合约时间戳。""" + storage_manager_mock.get_symbol_timestamp.return_value = { + "last_refresh_at": datetime(2026, 7, 4, 10, 0, 0).isoformat(), + "refresh_count": 42, + } + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + result = cache.get_symbol_timestamp(db, "AG2606", "futures") + + storage_manager_mock.get_symbol_timestamp.assert_called_once_with("AG2606", "futures") + assert result == datetime(2026, 7, 4, 10, 0, 0) + + def test_get_symbol_timestamp_falls_back_to_sqlite(self, db, storage_manager_mock): + """StorageManager 未命中时降级到 SQLite 读取合约时间戳。""" + storage_manager_mock.get_symbol_timestamp.return_value = None + _seed_sqlite_symbol_timestamp(db) + + with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock): + from app.services import cache + + result = cache.get_symbol_timestamp(db, "AG2606", "futures") + + assert result == datetime(2026, 7, 4, 10, 0, 0)