TIANHE
233 lines
6.2 KiB
Python
233 lines
6.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
===================================
|
||
数据缓存管理模块
|
||
===================================
|
||
|
||
参考 daily_stock_analysis 项目实现
|
||
用于缓存实时行情和K线数据，减少重复请求
|
||
|
||
特性：
|
||
1. TTL (Time To Live) 过期机制
|
||
2. LRU (Least Recently Used) 淘汰策略
|
||
3. 按数据类型分区管理
|
||
"""
|
||
|
||
import time
|
||
import logging
|
||
from typing import Dict, Any, Optional, List
|
||
from collections import OrderedDict
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
import threading
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class CacheEntry:
|
||
"""缓存条目"""
|
||
data: Any
|
||
timestamp: float
|
||
ttl: float
|
||
hit_count: int = 0
|
||
|
||
def is_expired(self) -> bool:
|
||
"""检查是否过期"""
|
||
return time.time() - self.timestamp > self.ttl
|
||
|
||
def age(self) -> float:
|
||
"""返回缓存年龄（秒）"""
|
||
return time.time() - self.timestamp
|
||
|
||
|
||
class DataCache:
|
||
"""
|
||
数据缓存管理器
|
||
|
||
特性：
|
||
- TTL 过期机制
|
||
- 最大容量限制
|
||
- LRU 淘汰策略
|
||
- 线程安全
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
name: str = "default",
|
||
default_ttl: float = 600.0, # 默认10分钟
|
||
max_size: int = 1000 # 最大缓存条目数
|
||
):
|
||
self.name = name
|
||
self.default_ttl = default_ttl
|
||
self.max_size = max_size
|
||
self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
|
||
self._lock = threading.RLock()
|
||
|
||
# 统计信息
|
||
self._hits = 0
|
||
self._misses = 0
|
||
|
||
def get(self, key: str) -> Optional[Any]:
|
||
"""
|
||
获取缓存数据
|
||
|
||
Returns:
|
||
缓存的数据，不存在或过期返回 None
|
||
"""
|
||
with self._lock:
|
||
if key not in self._cache:
|
||
self._misses += 1
|
||
return None
|
||
|
||
entry = self._cache[key]
|
||
|
||
# 检查是否过期
|
||
if entry.is_expired():
|
||
del self._cache[key]
|
||
self._misses += 1
|
||
logger.debug(f"[缓存] {self.name}:{key} 已过期，删除")
|
||
return None
|
||
|
||
# 更新访问顺序（LRU）
|
||
self._cache.move_to_end(key)
|
||
entry.hit_count += 1
|
||
self._hits += 1
|
||
|
||
logger.debug(f"[缓存命中] {self.name}:{key} (年龄: {entry.age():.0f}s/{entry.ttl:.0f}s)")
|
||
return entry.data
|
||
|
||
def set(
|
||
self,
|
||
key: str,
|
||
data: Any,
|
||
ttl: Optional[float] = None
|
||
) -> None:
|
||
"""
|
||
设置缓存数据
|
||
|
||
Args:
|
||
key: 缓存键
|
||
data: 缓存数据
|
||
ttl: 过期时间（秒），None 使用默认值
|
||
"""
|
||
with self._lock:
|
||
# 检查容量，执行 LRU 淘汰
|
||
while len(self._cache) >= self.max_size:
|
||
oldest_key, _ = self._cache.popitem(last=False)
|
||
logger.debug(f"[缓存] {self.name} 容量已满，淘汰: {oldest_key}")
|
||
|
||
actual_ttl = ttl if ttl is not None else self.default_ttl
|
||
self._cache[key] = CacheEntry(
|
||
data=data,
|
||
timestamp=time.time(),
|
||
ttl=actual_ttl
|
||
)
|
||
|
||
logger.debug(f"[缓存更新] {self.name}:{key} TTL={actual_ttl}s")
|
||
|
||
def delete(self, key: str) -> bool:
|
||
"""删除缓存条目"""
|
||
with self._lock:
|
||
if key in self._cache:
|
||
del self._cache[key]
|
||
logger.debug(f"[缓存] {self.name}:{key} 已删除")
|
||
return True
|
||
return False
|
||
|
||
def clear(self) -> int:
|
||
"""清空缓存"""
|
||
with self._lock:
|
||
count = len(self._cache)
|
||
self._cache.clear()
|
||
logger.info(f"[缓存] {self.name} 已清空 {count} 条记录")
|
||
return count
|
||
|
||
def cleanup_expired(self) -> int:
|
||
"""清理过期条目"""
|
||
with self._lock:
|
||
expired_keys = [
|
||
key for key, entry in self._cache.items()
|
||
if entry.is_expired()
|
||
]
|
||
for key in expired_keys:
|
||
del self._cache[key]
|
||
|
||
if expired_keys:
|
||
logger.debug(f"[缓存] {self.name} 清理 {len(expired_keys)} 条过期记录")
|
||
return len(expired_keys)
|
||
|
||
def stats(self) -> Dict[str, Any]:
|
||
"""获取缓存统计信息"""
|
||
with self._lock:
|
||
total_requests = self._hits + self._misses
|
||
hit_rate = self._hits / total_requests if total_requests > 0 else 0
|
||
|
||
return {
|
||
'name': self.name,
|
||
'size': len(self._cache),
|
||
'max_size': self.max_size,
|
||
'hits': self._hits,
|
||
'misses': self._misses,
|
||
'hit_rate': f"{hit_rate:.1%}",
|
||
'default_ttl': self.default_ttl
|
||
}
|
||
|
||
|
||
# ============================================
|
||
# 全局缓存实例
|
||
# ============================================
|
||
|
||
# 实时行情缓存（20分钟TTL）
|
||
_realtime_cache = DataCache(
|
||
name="realtime",
|
||
default_ttl=1200.0, # 20分钟
|
||
max_size=6000
|
||
)
|
||
|
||
# K线数据缓存（5分钟TTL，按需缓存）
|
||
_kline_cache = DataCache(
|
||
name="kline",
|
||
default_ttl=300.0, # 5分钟
|
||
max_size=500 # 最多500个交易对
|
||
)
|
||
|
||
# 股票基本信息缓存（1天TTL）
|
||
_stock_info_cache = DataCache(
|
||
name="stock_info",
|
||
default_ttl=86400.0, # 24小时
|
||
max_size=6000
|
||
)
|
||
|
||
|
||
def get_realtime_cache() -> DataCache:
|
||
"""获取实时行情缓存"""
|
||
return _realtime_cache
|
||
|
||
|
||
def get_kline_cache() -> DataCache:
|
||
"""获取K线数据缓存"""
|
||
return _kline_cache
|
||
|
||
|
||
def get_stock_info_cache() -> DataCache:
|
||
"""获取股票信息缓存"""
|
||
return _stock_info_cache
|
||
|
||
|
||
def generate_kline_cache_key(
|
||
symbol: str,
|
||
timeframe: str,
|
||
limit: int,
|
||
before_time: Optional[int] = None
|
||
) -> str:
|
||
"""
|
||
生成K线缓存键
|
||
|
||
格式: symbol:timeframe:limit[:before_time]
|
||
"""
|
||
key = f"{symbol}:{timeframe}:{limit}"
|
||
if before_time:
|
||
key += f":{before_time}"
|
||
return key
|