|
|
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
|
