This commit is contained in:
薇薇安 2026-01-18 09:07:48 +08:00
parent e41c4fe2e8
commit b8a77eb18d
5 changed files with 257 additions and 17 deletions

View File

@ -381,24 +381,24 @@ async def update_config(key: str, item: ConfigUpdate):
detail=f"{key} must be between 0 and 1 (0% to 100%)"
)
# 更新配置
# 更新配置会同时更新数据库和Redis缓存
TradingConfig.set(key, item.value, config_type, category, description)
# 清除配置缓存,确保立即生效
# 更新config_manager的缓存包括Redis
try:
from config_manager import ConfigManager
# 如果存在全局配置管理器实例,清除其缓存
import config_manager
if hasattr(config_manager, '_config_manager') and config_manager._config_manager:
config_manager._config_manager.reload()
if hasattr(config_manager, 'config_manager') and config_manager.config_manager:
# 调用set方法会同时更新数据库、Redis和本地缓存
config_manager.config_manager.set(key, item.value, config_type, category, description)
logger.info(f"配置已更新到Redis缓存: {key} = {item.value}")
except Exception as e:
logger.debug(f"清除配置缓存失败: {e}")
logger.warning(f"更新配置缓存失败: {e}")
return {
"message": "配置已更新",
"key": key,
"value": item.value,
"note": "交易系统将在下次扫描时自动使用新配置"
"note": "配置已同步到Redis交易系统将立即使用新配置"
}
except HTTPException:
raise

View File

@ -1,9 +1,12 @@
"""
配置管理器 - 从数据库读取配置兼容原有config.py
支持Redis缓存实现配置即时生效
"""
import sys
import os
import json
from pathlib import Path
from typing import Optional, Any
# 加载.env文件
try:
@ -42,22 +45,176 @@ import logging
logger = logging.getLogger(__name__)
# 尝试导入同步Redis客户端用于配置缓存
try:
import redis
REDIS_SYNC_AVAILABLE = True
except ImportError:
REDIS_SYNC_AVAILABLE = False
redis = None
class ConfigManager:
"""配置管理器 - 优先从数据库读取,回退到环境变量和默认值"""
"""配置管理器 - 优先从Redis缓存读取其次从数据库读取,回退到环境变量和默认值"""
def __init__(self):
self._cache = {}
self._redis_client: Optional[redis.Redis] = None
self._redis_connected = False
self._init_redis()
self._load_from_db()
def _init_redis(self):
"""初始化Redis客户端同步"""
if not REDIS_SYNC_AVAILABLE:
logger.debug("redis-py未安装配置缓存将不使用Redis")
return
try:
# 从环境变量或配置获取Redis连接信息
redis_url = os.getenv('REDIS_URL', 'redis://localhost:6379')
redis_use_tls = os.getenv('REDIS_USE_TLS', 'False').lower() == 'true'
redis_username = os.getenv('REDIS_USERNAME', None)
redis_password = os.getenv('REDIS_PASSWORD', None)
# 解析Redis URL
if redis_url.startswith('rediss://') or redis_use_tls:
# TLS连接
import ssl
ssl_context = ssl.create_default_context()
self._redis_client = redis.from_url(
redis_url,
username=redis_username,
password=redis_password,
ssl=ssl_context,
decode_responses=True
)
else:
# 普通连接
self._redis_client = redis.from_url(
redis_url,
username=redis_username,
password=redis_password,
decode_responses=True
)
# 测试连接
self._redis_client.ping()
self._redis_connected = True
logger.info("✓ Redis配置缓存连接成功")
except Exception as e:
logger.debug(f"Redis配置缓存连接失败: {e},将使用数据库缓存")
self._redis_client = None
self._redis_connected = False
def _get_from_redis(self, key: str) -> Optional[Any]:
"""从Redis获取配置值"""
if not self._redis_connected or not self._redis_client:
return None
try:
# 使用Hash存储所有配置键为 trading_config:{key}
value = self._redis_client.hget('trading_config', key)
if value:
# 尝试解析JSON如果是复杂类型
try:
return json.loads(value)
except (json.JSONDecodeError, TypeError):
return value
except Exception as e:
logger.debug(f"从Redis获取配置失败 {key}: {e}")
# 连接失败时,尝试重新连接
try:
self._redis_client.ping()
self._redis_connected = True
except:
self._redis_connected = False
return None
def _set_to_redis(self, key: str, value: Any):
"""设置配置到Redis"""
if not self._redis_connected or not self._redis_client:
return False
try:
# 使用Hash存储所有配置键为 trading_config:{key}
# 将值序列化为JSON如果是复杂类型
if isinstance(value, (dict, list)):
value_str = json.dumps(value)
else:
value_str = str(value)
self._redis_client.hset('trading_config', key, value_str)
# 设置整个Hash的过期时间为7天配置不会频繁变化但需要定期刷新
self._redis_client.expire('trading_config', 7 * 24 * 3600)
return True
except Exception as e:
logger.debug(f"设置配置到Redis失败 {key}: {e}")
# 连接失败时,尝试重新连接
try:
self._redis_client.ping()
self._redis_connected = True
# 重试一次
if isinstance(value, (dict, list)):
value_str = json.dumps(value)
else:
value_str = str(value)
self._redis_client.hset('trading_config', key, value_str)
self._redis_client.expire('trading_config', 7 * 24 * 3600)
return True
except:
self._redis_connected = False
return False
def _load_all_to_redis(self):
"""将所有配置加载到Redis"""
if not self._redis_connected or not self._redis_client:
return
try:
# 批量设置所有配置到Redis
pipe = self._redis_client.pipeline()
for key, value in self._cache.items():
if isinstance(value, (dict, list)):
value_str = json.dumps(value)
else:
value_str = str(value)
pipe.hset('trading_config', key, value_str)
pipe.expire('trading_config', 7 * 24 * 3600)
pipe.execute()
logger.debug(f"已将 {len(self._cache)} 个配置项同步到Redis")
except Exception as e:
logger.debug(f"同步配置到Redis失败: {e}")
def _load_from_db(self):
"""从数据库加载配置"""
"""从数据库加载配置如果Redis中没有"""
if TradingConfig is None:
logger.warning("TradingConfig未导入无法从数据库加载配置")
self._cache = {}
return
try:
# 先尝试从Redis加载所有配置
if self._redis_connected and self._redis_client:
try:
redis_configs = self._redis_client.hgetall('trading_config')
if redis_configs:
# 解析Redis中的配置
for key, value_str in redis_configs.items():
try:
# 尝试解析JSON
value = json.loads(value_str)
except (json.JSONDecodeError, TypeError):
# 如果不是JSON尝试转换类型
value = value_str
self._cache[key] = value
logger.info(f"从Redis加载了 {len(self._cache)} 个配置项")
return
except Exception as e:
logger.debug(f"从Redis加载配置失败: {e},回退到数据库")
# 从数据库加载配置
configs = TradingConfig.get_all()
for config in configs:
key = config['config_key']
@ -66,45 +223,85 @@ class ConfigManager:
config['config_type']
)
self._cache[key] = value
logger.info(f"从数据库加载了 {len(self._cache)} 个配置项")
# 同时写入Redis缓存
self._set_to_redis(key, value)
logger.info(f"从数据库加载了 {len(self._cache)} 个配置项已同步到Redis")
except Exception as e:
logger.warning(f"从数据库加载配置失败,使用默认配置: {e}")
self._cache = {}
def get(self, key, default=None):
"""获取配置值"""
# 1. 优先从数据库缓存读取
# 1. 优先从Redis缓存读取最新
redis_value = self._get_from_redis(key)
if redis_value is not None:
# 同时更新本地缓存
self._cache[key] = redis_value
return redis_value
# 2. 从本地缓存读取
if key in self._cache:
return self._cache[key]
# 2. 从环境变量读取
# 3. 从环境变量读取
env_value = os.getenv(key)
if env_value is not None:
return env_value
# 3. 返回默认值
# 4. 返回默认值
return default
def set(self, key, value, config_type='string', category='general', description=None):
"""设置配置(同时更新数据库和缓存)"""
"""设置配置(同时更新数据库、Redis缓存本地缓存)"""
if TradingConfig is None:
logger.warning("TradingConfig未导入无法更新数据库配置")
self._cache[key] = value
# 仍然尝试更新Redis
self._set_to_redis(key, value)
return
try:
# 1. 更新数据库
TradingConfig.set(key, value, config_type, category, description)
# 2. 更新本地缓存
self._cache[key] = value
logger.info(f"配置已更新: {key} = {value}")
# 3. 更新Redis缓存确保trading_system能立即读取到最新配置
self._set_to_redis(key, value)
logger.info(f"配置已更新: {key} = {value} (已同步到数据库和Redis)")
except Exception as e:
logger.error(f"更新配置失败: {e}")
raise
def reload(self):
"""重新加载配置"""
"""重新加载配置优先从Redis其次从数据库"""
self._cache = {}
self._load_from_db()
def reload_from_redis(self):
"""强制从Redis重新加载配置用于trading_system即时获取最新配置"""
self._cache = {}
if self._redis_connected and self._redis_client:
try:
redis_configs = self._redis_client.hgetall('trading_config')
if redis_configs:
for key, value_str in redis_configs.items():
try:
value = json.loads(value_str)
except (json.JSONDecodeError, TypeError):
value = value_str
self._cache[key] = value
logger.debug(f"从Redis重新加载了 {len(self._cache)} 个配置项")
return
except Exception as e:
logger.debug(f"从Redis重新加载配置失败: {e}")
# 如果Redis失败回退到数据库
self._load_from_db()
def get_trading_config(self):
"""获取交易配置字典兼容原有config.py的TRADING_CONFIG"""
return {

View File

@ -165,9 +165,16 @@ def _get_config_value(key, default=None):
return default
def _get_trading_config():
<<<<<<< Current (Your changes)
"""获取交易配置(支持动态重载)"""
if _config_manager:
_config_manager.reload() # 每次获取配置时重新加载
=======
"""获取交易配置支持动态重载优先从Redis读取最新值"""
if _config_manager:
# 从Redis重新加载配置轻量级只从Redis读取不查数据库
_config_manager.reload_from_redis()
>>>>>>> Incoming (Background Agent changes)
return _config_manager.get_trading_config()
# 回退到默认配置
return {

View File

@ -155,6 +155,16 @@ class PositionManager:
logger.debug(f"获取K线数据失败使用固定止损: {e}")
klines = None
# 在开仓前从Redis重新加载配置确保使用最新配置包括ATR参数
# 从Redis读取最新配置轻量级即时生效
try:
if config._config_manager:
config._config_manager.reload_from_redis()
config.TRADING_CONFIG = config._get_trading_config()
logger.debug(f"{symbol} 开仓前已从Redis重新加载配置")
except Exception as e:
logger.debug(f"从Redis重新加载配置失败: {e}")
# 使用基于保证金的止损止盈(结合技术分析)
# 计算保证金和仓位价值
position_value = entry_price * quantity
@ -769,6 +779,14 @@ class PositionManager:
position_info['maxProfit'] = pnl_percent_margin
# 移动止损逻辑(盈利后保护利润,基于保证金)
# 每次检查时从Redis重新加载配置确保配置修改能即时生效
try:
if config._config_manager:
config._config_manager.reload_from_redis()
config.TRADING_CONFIG = config._get_trading_config()
except Exception as e:
logger.debug(f"从Redis重新加载配置失败: {e}")
use_trailing = config.TRADING_CONFIG.get('USE_TRAILING_STOP', True)
if use_trailing:
trailing_activation = config.TRADING_CONFIG.get('TRAILING_STOP_ACTIVATION', 0.01) # 相对于保证金
@ -1781,6 +1799,14 @@ class PositionManager:
position_info['maxProfit'] = pnl_percent_margin
# 移动止损逻辑(盈利后保护利润,基于保证金)
# 每次检查时从Redis重新加载配置确保配置修改能即时生效
try:
if config._config_manager:
config._config_manager.reload_from_redis()
config.TRADING_CONFIG = config._get_trading_config()
except Exception as e:
logger.debug(f"从Redis重新加载配置失败: {e}")
use_trailing = config.TRADING_CONFIG.get('USE_TRAILING_STOP', True)
if use_trailing:
trailing_activation = config.TRADING_CONFIG.get('TRAILING_STOP_ACTIVATION', 0.01) # 相对于保证金

View File

@ -71,6 +71,16 @@ class TradingStrategy:
try:
while self.running:
# 0. 定期从Redis重新加载配置确保配置修改能即时生效
# 每次循环开始时从Redis重新加载确保使用最新配置
try:
if config._config_manager:
config._config_manager.reload_from_redis()
config.TRADING_CONFIG = config._get_trading_config()
logger.debug("配置已从Redis重新加载")
except Exception as e:
logger.warning(f"从Redis重新加载配置失败: {e}")
# 1. 扫描市场找出涨跌幅最大的前N个货币对
top_symbols = await self.scanner.scan_market()