import datetime as _datetime import json import logging import time from datetime import datetime 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 _parse_fetched_at(self, timestamp_raw): """优先使用输入 timestamp,解析失败时回退到当前时间。""" if not timestamp_raw: return _datetime.datetime.now() if isinstance(timestamp_raw, str): try: return _datetime.datetime.fromisoformat(timestamp_raw) except Exception: return _datetime.datetime.now() if hasattr(timestamp_raw, "isoformat"): return timestamp_raw return _datetime.datetime.now() 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 def save_market_data(self, symbol, data): """ 保存行情数据:先删 Redis 缓存,再写 MySQL,MySQL 成功后回填 Redis。 Args: symbol: 品种合约代码 data: 采集脚本返回的数据,包含 timeframes, current_price, timestamp, type Returns: bool: MySQL 写入是否成功 """ data_type = data.get("type", "futures") timeframes = data.get("timeframes", {}) current_price = data.get("current_price") fetched_at = self._parse_fetched_at(data.get("timestamp")) cache_keys = [f"market_data:{symbol}:{period}" for period in timeframes.keys()] # 1. 删除 Redis 缓存 if self.check_redis() and cache_keys: try: self.redis_client.delete(*cache_keys) except Exception as e: logger.warning(f"Redis 删除缓存失败 [{symbol}]: {e}") # 2. 写入 MySQL(事务) if not self.check_mysql(): logger.warning(f"MySQL 不可用,无法写入行情数据 [{symbol}]") return False try: SessionLocal = sessionmaker(bind=self.mysql_engine) with SessionLocal() as db: try: for period, candles in timeframes.items(): candles_json = json.dumps(candles, ensure_ascii=False) record = ( db.query(MarketData) .filter_by(symbol=symbol, data_type=data_type, period=period) .first() ) if record: record.candles_json = candles_json record.current_price = current_price record.fetched_at = fetched_at record.candle_count = len(candles) else: record = MarketData( symbol=symbol, data_type=data_type, period=period, candles_json=candles_json, current_price=current_price, fetched_at=fetched_at, candle_count=len(candles), ) db.add(record) db.commit() except Exception: db.rollback() raise except Exception as e: logger.error(f"MySQL 写入行情数据失败 [{symbol}]: {e}") return False # 3. MySQL 成功,回填 Redis(失败仅记录日志) if self.check_redis(): for period, candles in timeframes.items(): cache_key = f"market_data:{symbol}:{period}" cache_value = { "current_price": current_price, "timestamp": fetched_at.isoformat(), "candles": candles, } try: self.redis_client.setex( cache_key, CACHE_TTL_SECONDS, json.dumps(cache_value, ensure_ascii=False), ) except Exception as e: logger.warning(f"Redis 更新失败 [{cache_key}]: {e}") return True def save_symbol_timestamp(self, symbol, data_type, refresh_time): """ 保存合约时间戳:先删 Redis 缓存,再写 MySQL,MySQL 成功后回填 Redis。 Args: symbol: 品种合约代码 data_type: 数据类型,如 futures refresh_time: 刷新时间(datetime) Returns: bool: MySQL 写入是否成功 """ cache_key = f"symbol_timestamps:{symbol}" # 1. 删除 Redis 缓存 if self.check_redis(): try: self.redis_client.delete(cache_key) except Exception as e: logger.warning(f"Redis 删除缓存失败 [{cache_key}]: {e}") # 2. 写入 MySQL(事务) if not self.check_mysql(): logger.warning(f"MySQL 不可用,无法写入合约时间戳 [{symbol}]") return False try: SessionLocal = sessionmaker(bind=self.mysql_engine) with SessionLocal() as db: try: record = ( db.query(SymbolTimestamp) .filter_by(symbol=symbol, data_type=data_type) .first() ) if record: record.last_refresh_at = refresh_time record.refresh_count += 1 else: record = SymbolTimestamp( symbol=symbol, data_type=data_type, last_refresh_at=refresh_time, refresh_count=1, ) db.add(record) db.commit() except Exception: db.rollback() raise except Exception as e: logger.error(f"MySQL 写入合约时间戳失败 [{symbol}]: {e}") return False # 3. MySQL 成功,回填 Redis(失败仅记录日志) if self.check_redis(): cache_value = { "last_refresh_at": refresh_time.isoformat(), "refresh_count": record.refresh_count, } try: self.redis_client.setex( cache_key, CACHE_TTL_SECONDS, json.dumps(cache_value, ensure_ascii=False), ) except Exception as e: logger.warning(f"Redis 更新失败 [{cache_key}]: {e}") return True def delete_cache(self, symbol, periods): """ 批量删除 Redis 行情数据缓存键。 Args: symbol: 品种合约代码 periods: 周期列表,如 ["5min", "15min"] """ if not self.check_redis(): return cache_keys = [f"market_data:{symbol}:{period}" for period in periods] if not cache_keys: return try: self.redis_client.delete(*cache_keys) except Exception as e: logger.warning(f"Redis 批量删除缓存失败 [{symbol}]: {e}") _storage_manager = None def get_storage_manager(): """获取全局 StorageManager 单例。""" global _storage_manager if _storage_manager is None: _storage_manager = StorageManager() return _storage_manager