import json import logging import time from sqlalchemy import text from sqlalchemy.orm import sessionmaker from app.models import MarketData, SymbolTimestamp logger = logging.getLogger(__name__) # Redis 缓存 TTL:30 天 CACHE_TTL_SECONDS = 30 * 24 * 60 * 60 class StorageManager: """存储管理器骨架,封装 Redis / MySQL 可用性检测与三级降级状态。""" def __init__(self, redis_client=None, mysql_engine=None, check_interval=30): self.redis_client = redis_client self.mysql_engine = mysql_engine self.check_interval = check_interval self.redis_available = False self.mysql_available = False self.last_redis_check = 0 self.last_mysql_check = 0 def check_redis(self): """惰性检测 Redis 可用性,30 秒间隔内返回缓存结果。""" now = time.time() if now - self.last_redis_check < self.check_interval: return self.redis_available if self.redis_client is None: self.redis_available = False else: try: self.redis_client.ping() self.redis_available = True except Exception as e: self.redis_available = False self.last_redis_check = now return self.redis_available def check_mysql(self): """惰性检测 MySQL 可用性,30 秒间隔内返回缓存结果。""" now = time.time() if now - self.last_mysql_check < self.check_interval: return self.mysql_available if self.mysql_engine is None: self.mysql_available = False else: try: with self.mysql_engine.connect() as conn: conn.execute(text("SELECT 1")) self.mysql_available = True except Exception as e: self.mysql_available = False self.last_mysql_check = now return self.mysql_available def is_available(self): """判断 Redis 或 MySQL 至少一个可用。""" return self.check_redis() or self.check_mysql() def initialize(self): """启动时调用,检测各后端可用性并输出日志。""" redis_ok = self.check_redis() mysql_ok = self.check_mysql() if redis_ok: logger.info("Redis 可用") else: logger.warning("Redis 不可用") if mysql_ok: logger.info("MySQL 可用") else: logger.warning("MySQL 不可用") def get_market_data(self, symbol, data_type, periods): """ 读取行情数据:Redis → MySQL → None(由调用方降级到 SQLite)。 Args: symbol: 品种合约代码 data_type: 数据类型,如 futures periods: 周期,如 5min(对应 Redis key 中的 period) Returns: dict 或 None """ cache_key = f"market_data:{symbol}:{periods}" # 1. 检查 Redis 缓存 if self.check_redis(): try: raw = self.redis_client.get(cache_key) if raw: return json.loads(raw) except Exception as e: logger.warning(f"Redis 读取失败 [{cache_key}]: {e}") # 2. Redis 未命中,检查 MySQL 可用性 if not self.check_mysql(): return None # 3. 从 MySQL 读取 try: SessionLocal = sessionmaker(bind=self.mysql_engine) with SessionLocal() as db: record = ( db.query(MarketData) .filter_by(symbol=symbol, data_type=data_type, period=periods) .first() ) if not record: return None result = { "current_price": record.current_price, "timestamp": ( record.fetched_at.isoformat() if record.fetched_at else None ), "candles": ( json.loads(record.candles_json) if record.candles_json else [] ), } # 4. 回填 Redis(TTL 30 天) if self.check_redis(): try: self.redis_client.setex( cache_key, CACHE_TTL_SECONDS, json.dumps(result, ensure_ascii=False), ) except Exception as e: logger.warning(f"Redis 回填失败 [{cache_key}]: {e}") return result except Exception as e: logger.warning(f"MySQL 读取失败 [{cache_key}]: {e}") return None def get_symbol_timestamp(self, symbol, data_type): """ 读取合约时间戳:Redis → MySQL → None(由调用方降级到 SQLite)。 Args: symbol: 品种合约代码 data_type: 数据类型,如 futures Returns: dict 或 None """ cache_key = f"symbol_timestamps:{symbol}" # 1. 检查 Redis 缓存 if self.check_redis(): try: raw = self.redis_client.get(cache_key) if raw: return json.loads(raw) except Exception as e: logger.warning(f"Redis 读取失败 [{cache_key}]: {e}") # 2. Redis 未命中,检查 MySQL 可用性 if not self.check_mysql(): return None # 3. 从 MySQL 读取 try: SessionLocal = sessionmaker(bind=self.mysql_engine) with SessionLocal() as db: record = ( db.query(SymbolTimestamp) .filter_by(symbol=symbol, data_type=data_type) .first() ) if not record: return None result = { "last_refresh_at": ( record.last_refresh_at.isoformat() if record.last_refresh_at else None ), "refresh_count": record.refresh_count, } # 4. 回填 Redis(TTL 30 天) if self.check_redis(): try: self.redis_client.setex( cache_key, CACHE_TTL_SECONDS, json.dumps(result, ensure_ascii=False), ) except Exception as e: logger.warning(f"Redis 回填失败 [{cache_key}]: {e}") return result except Exception as e: logger.warning(f"MySQL 读取失败 [{cache_key}]: {e}") return None _storage_manager = None def get_storage_manager(): """获取全局 StorageManager 单例。""" global _storage_manager if _storage_manager is None: _storage_manager = StorageManager() return _storage_manager