auto_trade_sys/trading_system/config.py
薇薇安 07d3bf4398 a
2026-01-25 16:31:37 +08:00

391 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
配置文件 - API密钥和交易参数配置
支持从数据库读取配置(优先),回退到环境变量和默认值
支持动态重载配置
"""
import os
from typing import Optional
from pathlib import Path
# 加载 .env 文件(优先从 trading_system/.env其次从 backend/.env再到项目根目录/.env
# 重要:对于 ATS_ACCOUNT_ID如果 supervisor 已设置(交易进程),则不要从 .env 覆盖
# .env 中的 ATS_ACCOUNT_ID 只用于 recommendations_main.py 的默认值
try:
from dotenv import load_dotenv
trading_system_dir = Path(__file__).parent
project_root = trading_system_dir.parent
backend_dir = project_root / "backend"
env_files = [
trading_system_dir / '.env', # trading_system/.env
backend_dir / '.env', # backend/.env线上常见放这里DB/REDIS/ATS_MASTER_KEY
project_root / '.env', # 项目根目录/.env
]
# 先保存 supervisor 设置的 ATS_ACCOUNT_ID如果存在
# supervisor 在启动交易进程时会设置这个环境变量,优先级应该高于 .env
supervisor_account_id = os.getenv("ATS_ACCOUNT_ID")
for env_file in env_files:
if env_file.exists():
# 加载 .env 文件(会覆盖环境变量)
load_dotenv(env_file, override=True)
print(f"[config.py] 已加载 .env 文件: {env_file}")
# 如果 supervisor 设置了 ATS_ACCOUNT_ID恢复它交易进程使用 supervisor 的值,不依赖 .env
if supervisor_account_id:
os.environ["ATS_ACCOUNT_ID"] = supervisor_account_id
print(f"[config.py] 恢复 supervisor 设置的 ATS_ACCOUNT_ID={supervisor_account_id}(交易进程使用此值,不依赖 .env")
break
else:
raise Exception("supervisor 未设置 ATS_ACCOUNT_ID")
# # 如果都不存在,尝试加载但不报错
# load_dotenv(project_root / '.env', override=False)
# # 同样恢复 supervisor 设置的值
# if supervisor_account_id:
# os.environ["ATS_ACCOUNT_ID"] = supervisor_account_id
except ImportError:
# python-dotenv 未安装时忽略
pass
except Exception as e:
# 加载 .env 文件失败时忽略,不影响程序运行
print(f"[config.py] 加载ATS_ACCOUNT_ID时出错: {e}")
# 尝试从数据库加载配置
USE_DB_CONFIG = False
_config_manager = None
def _init_config_manager():
"""初始化配置管理器"""
global USE_DB_CONFIG, _config_manager
if _config_manager is not None:
return _config_manager
# 使用基础日志(因为可能还没有配置好日志系统)
import sys
from pathlib import Path
try:
# 从trading_system目录向上两级到项目根目录然后找backend
project_root = Path(__file__).parent.parent
backend_path = project_root / 'backend'
print(f"[配置管理器] 尝试初始化,项目根目录: {project_root}")
print(f"[配置管理器] backend路径: {backend_path}")
print(f"[配置管理器] backend目录存在: {backend_path.exists()}")
if not backend_path.exists():
print(f"[配置管理器] ❌ backend目录不存在: {backend_path}")
USE_DB_CONFIG = False
return None
# 检查backend目录内容
backend_files = list(backend_path.iterdir()) if backend_path.exists() else []
print(f"[配置管理器] backend目录内容: {[f.name for f in backend_files[:10]]}")
# 检查config_manager.py是否存在
config_manager_file = backend_path / 'config_manager.py'
print(f"[配置管理器] config_manager.py存在: {config_manager_file.exists()}")
if not config_manager_file.exists():
print(f"[配置管理器] ❌ config_manager.py不存在: {config_manager_file}")
USE_DB_CONFIG = False
return None
# 添加到sys.path
backend_str = str(backend_path)
if backend_str not in sys.path:
sys.path.insert(0, backend_str)
print(f"[配置管理器] 已添加路径到sys.path: {backend_str}")
# 尝试导入
try:
print("[配置管理器] 尝试导入config_manager...")
from config_manager import ConfigManager # type: ignore
print("[配置管理器] ✓ 导入成功")
# 从环境变量获取 account_idtrading_system 进程通过 ATS_ACCOUNT_ID 指定)
try:
account_id = int(os.getenv("ATS_ACCOUNT_ID") or os.getenv("ACCOUNT_ID") or 1)
except Exception:
account_id = 1
print(f"[配置管理器] 使用 account_id={account_id} (从环境变量 ATS_ACCOUNT_ID 读取)")
# 为当前账号创建 ConfigManager 实例(而不是使用全局的)
config_manager = ConfigManager.for_account(account_id)
print(f"[配置管理器] ✓ 已创建 ConfigManager 实例 (account_id={account_id})")
# 测试数据库连接
try:
print("[配置管理器] 测试数据库连接...")
config_manager.reload()
print(f"[配置管理器] ✓ 数据库连接成功,已加载 {len(config_manager._cache)} 个配置项")
# 检查API密钥使用正确的 account_id
api_key = config_manager.get('BINANCE_API_KEY')
api_secret = config_manager.get('BINANCE_API_SECRET')
print(f"[配置管理器] API密钥检查: KEY存在={bool(api_key)}, SECRET存在={bool(api_secret)}")
if api_key and api_key != 'your_api_key_here':
print(f"[配置管理器] API_KEY前4位: {api_key[:4]}...")
_config_manager = config_manager
USE_DB_CONFIG = True
print("[配置管理器] ✓ 配置管理器初始化成功,将从数据库读取配置")
return config_manager
except Exception as db_error:
print(f"[配置管理器] ⚠ 数据库连接失败: {db_error}")
print(f"[配置管理器] 错误类型: {type(db_error).__name__}")
import traceback
print(f"[配置管理器] 错误详情:\n{traceback.format_exc()}")
USE_DB_CONFIG = False
return None
except ImportError as e:
print(f"[配置管理器] ❌ 无法导入config_manager: {e}")
import traceback
print(f"[配置管理器] 导入错误详情:\n{traceback.format_exc()}")
USE_DB_CONFIG = False
return None
except Exception as e:
print(f"[配置管理器] ❌ 配置管理器初始化失败: {e}")
import traceback
print(f"[配置管理器] 错误详情:\n{traceback.format_exc()}")
USE_DB_CONFIG = False
return None
except Exception as e:
print(f"[配置管理器] ❌ 初始化异常: {e}")
import traceback
print(f"[配置管理器] 异常详情:\n{traceback.format_exc()}")
USE_DB_CONFIG = False
return None
# 初始化配置管理器(在模块加载时执行)
# 注意此时日志系统可能还没初始化所以使用print输出
try:
_init_config_manager()
except Exception as e:
print(f"[config.py] 配置管理器初始化异常: {e}")
import traceback
print(traceback.format_exc())
def _get_config_value(key, default=None):
"""获取配置值(支持动态重载)"""
if _config_manager:
try:
# 不每次都reload避免性能问题只在需要时reload
value = _config_manager.get(key)
# 如果从数据库获取到值直接返回排除None和空字符串除非是敏感配置
if value is not None:
# 对于API密钥即使是空字符串也返回表示数据库中有配置但值为空
if key in ['BINANCE_API_KEY', 'BINANCE_API_SECRET']:
return value if value else default
# 对于其他配置空字符串视为无效返回default
if value == '':
return default
return value
except Exception as e:
import logging
logger = logging.getLogger(__name__)
logger.debug(f"从配置管理器获取{key}失败: {e},尝试环境变量")
# 回退到环境变量
env_value = os.getenv(key)
if env_value is not None and env_value != '':
return env_value
# 最后返回默认值
return default
def _get_trading_config():
"""获取交易配置支持动态重载优先从Redis读取最新值"""
if _config_manager:
# 从Redis重新加载配置轻量级只从Redis读取不查数据库
# 注意如果Redis不可用或没有数据reload_from_redis()会保持现有缓存,不会触发数据库加载
_config_manager.reload_from_redis()
return _config_manager.get_trading_config()
# 回退到默认配置
return {
# ===== 用户风险旋钮(山寨币专属策略)=====
'AUTO_TRADE_ENABLED': True, # 自动交易总开关
'MAX_OPEN_POSITIONS': 4, # 同时持仓数量上限总仓位12% / 单笔1.5% = 最多4个
'MAX_DAILY_ENTRIES': 5, # 每日最多5笔山寨币少做多看
'MAX_POSITION_PERCENT': 0.015, # 单笔仓位1.5%(山寨币风险高,不加仓)
'MAX_TOTAL_POSITION_PERCENT': 0.12, # 总仓位12%(保守控制总风险)
'MIN_POSITION_PERCENT': 0.01, # 最小仓位1%
'MIN_MARGIN_USDT': 5.0, # 最小保证金5美元
'MIN_CHANGE_PERCENT': 0.5, # 最小价格变动0.5%
'TOP_N_SYMBOLS': 8, # 选择信号最强的8个给更多选择余地避免错过好机会
'MAX_SCAN_SYMBOLS': 250, # 扫描前250个增加覆盖率从27.6%提升到46.0%
'EXCLUDE_MAJOR_COINS': True, # 排除主流币BTC、ETH、BNB等专注于山寨币
'STOP_LOSS_PERCENT': 0.15, # 止损15%(山寨币波动大,止损要宽)
'TAKE_PROFIT_PERCENT': 0.60, # 止盈60%4:1盈亏比追求大赢家
'MIN_STOP_LOSS_PRICE_PCT': 0.02, # 最小止损价格变动2%
'MIN_TAKE_PROFIT_PRICE_PCT': 0.02, # 最小止盈价格变动2%
'USE_ATR_STOP_LOSS': True, # 使用ATR动态止损
'ATR_STOP_LOSS_MULTIPLIER': 2.0, # ATR止损倍数2.0(容忍山寨币高波动)
'ATR_TAKE_PROFIT_MULTIPLIER': 8.0, # ATR止盈倍数8.0盈亏比4:1
'RISK_REWARD_RATIO': 4.0, # 盈亏比4:1山寨币必须追求大赢家
'ATR_PERIOD': 14, # ATR计算周期14
'USE_DYNAMIC_ATR_MULTIPLIER': False, # 不使用动态ATR
'ATR_MULTIPLIER_MIN': 1.5, # 动态ATR倍数最小值
'ATR_MULTIPLIER_MAX': 2.5, # 动态ATR倍数最大值
'SCAN_INTERVAL': 3600, # 扫描间隔1小时3600秒不要太频繁
'KLINE_INTERVAL': '1h',
'PRIMARY_INTERVAL': '4h', # 主周期4小时过滤噪音
'CONFIRM_INTERVAL': '1d', # 确认周期日线,看大趋势
'ENTRY_INTERVAL': '1h', # 入场周期1小时避免太小的时间框架
'MIN_VOLUME_24H': 30000000, # 24小时成交额≥3000万美元过滤垃圾币
'MIN_VOLUME_24H_STRICT': 50000000, # 严格过滤≥5000万美元
'MIN_VOLATILITY': 0.03, # 最小波动率3%,过滤死币
'MIN_SIGNAL_STRENGTH': 7, # 信号强度≥7保持较高门槛
# ===== 动态过滤优化 =====
'BETA_FILTER_ENABLED': True, # 大盘共振过滤BTC/ETH下跌时屏蔽多单
'BETA_FILTER_THRESHOLD': -0.03, # -3%BTC/ETH下跌超过此阈值时屏蔽多单
'ATR_SPIKE_THRESHOLD': 2.0, # ATR异常激增阈值当前ATR / 平均ATR
'SIGNAL_STRENGTH_POSITION_MULTIPLIER': {8: 0.5, 9: 1.0, 10: 1.0}, # 信号强度分级8分50%仓位9-10分100%仓位
# ===== 仓位管理优化(山寨币专属)=====
'USE_FIXED_RISK_SIZING': True, # 固定每笔风险,避免亏损扩大
'FIXED_RISK_PERCENT': 0.01, # 每笔最多亏总资金1%(山寨币风险高)
'MAX_LEVERAGE_SMALL_CAP': 8, # 小众币最大杠杆8倍
'ATR_LEVERAGE_REDUCTION_THRESHOLD': 0.05, # ATR超过5%时降低杠杆
'LEVERAGE': 8, # 基础杠杆降到8倍山寨币波动大
'USE_DYNAMIC_LEVERAGE': False, # 不使用动态杠杆(保持简单)
'MAX_LEVERAGE': 12, # 最大杠杆12倍不要超过
# 移动止损:必须开启!山寨币利润要保护
'USE_TRAILING_STOP': True,
'TRAILING_STOP_ACTIVATION': 0.30, # 盈利30%后激活(山寨币波动大)
'TRAILING_STOP_PROTECT': 0.15, # 保护15%利润(给回撤足够空间)
# 最小持仓时间锁立即取消山寨币30分钟可能暴涨暴跌50%
'MIN_HOLD_TIME_SEC': 0, # 取消持仓时间锁
'POSITION_SYNC_INTERVAL': 60, # 持仓状态同步间隔60秒
# ===== 自动交易过滤(用于提升胜率/控频)=====
# 是否仅在 marketRegime=trending 时才自动交易;否则只生成推荐
'AUTO_TRADE_ONLY_TRENDING': True,
# 是否允许 4H 趋势为 neutral 时自动交易;默认不允许(震荡最易扫损)
'AUTO_TRADE_ALLOW_4H_NEUTRAL': False,
# ===== 智能入场方案C=====
# 根治方案:默认关闭。关闭后回归“纯限价单模式”(不追价/不市价兜底/未成交撤单跳过)
'SMART_ENTRY_ENABLED': True, # 开启智能入场,提高成交率
'SMART_ENTRY_STRONG_SIGNAL': 7, # 强信号阈值≥7
'ENTRY_SYMBOL_COOLDOWN_SEC': 1800, # 同一币种冷却30分钟1800秒避免频繁操作
'ENTRY_TIMEOUT_SEC': 180, # 智能入场总预算(秒)(限价/追价逻辑内部使用)
'ENTRY_STEP_WAIT_SEC': 15, # 每步等待成交时间(秒)
'ENTRY_CHASE_MAX_STEPS': 4, # 最多追价步数(逐步减少 offset
'ENTRY_MARKET_FALLBACK_AFTER_SEC': 45, # 趋势强时:超过该秒数仍未成交 -> 评估是否市价兜底
'ENTRY_CONFIRM_TIMEOUT_SEC': 30, # 下单后最终确认成交的等待时间(秒)
'ENTRY_MAX_DRIFT_PCT_TRENDING': 0.8, # 追价偏离放宽到0.8%(山寨币跳空大)
'ENTRY_MAX_DRIFT_PCT_RANGING': 0.3, # 震荡/弱趋势时允许的最大追价偏离
}
# 币安API配置优先从数据库回退到环境变量和默认值
# 注意:在模块加载时,配置管理器可能还未初始化完成,所以先使用默认值
# 在main.py中会调用reload_config()重新加载
BINANCE_API_KEY: Optional[str] = _get_config_value('BINANCE_API_KEY', 'your_api_key_here')
BINANCE_API_SECRET: Optional[str] = _get_config_value('BINANCE_API_SECRET', 'your_api_secret_here')
USE_TESTNET: bool = _get_config_value('USE_TESTNET', False) if _get_config_value('USE_TESTNET') is not None else os.getenv('USE_TESTNET', 'False').lower() == 'true'
# 交易参数配置(优先从数据库读取,支持动态重载)
TRADING_CONFIG = _get_trading_config()
# 确保包含所有必要的默认值
defaults = {
# 用户风险旋钮即使DB里没配置也能用
'AUTO_TRADE_ENABLED': True,
'MAX_OPEN_POSITIONS': 3,
'MAX_DAILY_ENTRIES': 8,
'SCAN_INTERVAL': 1800,
'KLINE_INTERVAL': '1h',
'PRIMARY_INTERVAL': '1h',
'CONFIRM_INTERVAL': '4h',
'ENTRY_INTERVAL': '15m',
'LIMIT_ORDER_OFFSET_PCT': 0.5, # 限价单偏移百分比默认0.5%
# 智能入场默认值即使DB里没配置也能用
'SMART_ENTRY_ENABLED': False,
'ENTRY_SYMBOL_COOLDOWN_SEC': 120,
'ENTRY_TIMEOUT_SEC': 180,
'ENTRY_STEP_WAIT_SEC': 15,
'ENTRY_CHASE_MAX_STEPS': 4,
'ENTRY_MARKET_FALLBACK_AFTER_SEC': 45,
'ENTRY_CONFIRM_TIMEOUT_SEC': 30,
'ENTRY_MAX_DRIFT_PCT_TRENDING': 0.6,
'ENTRY_MAX_DRIFT_PCT_RANGING': 0.3,
# 自动交易过滤默认值
'AUTO_TRADE_ONLY_TRENDING': True,
'AUTO_TRADE_ALLOW_4H_NEUTRAL': False,
# 最小持仓时间锁(强制波段持仓纪律,避免分钟级平仓)
'MIN_HOLD_TIME_SEC': 1800, # 默认30分钟1800秒
}
for key, value in defaults.items():
if key not in TRADING_CONFIG:
TRADING_CONFIG[key] = value
# 提供一个函数来重新加载配置
def reload_config():
"""重新加载配置(供外部调用)"""
global TRADING_CONFIG, BINANCE_API_KEY, BINANCE_API_SECRET, USE_TESTNET, _config_manager, USE_DB_CONFIG
global REDIS_URL, REDIS_USE_TLS, REDIS_SSL_CERT_REQS, REDIS_SSL_CA_CERTS, REDIS_USERNAME, REDIS_PASSWORD, REDIS_SELECT
# 如果配置管理器不存在,尝试初始化
if _config_manager is None:
_init_config_manager()
# 如果配置管理器存在,重新加载
if _config_manager:
try:
_config_manager.reload()
USE_DB_CONFIG = True
except Exception as e:
import logging
logger = logging.getLogger(__name__)
logger.warning(f"重新加载配置失败: {e}")
USE_DB_CONFIG = False
# 重新获取API密钥优先从配置管理器
new_api_key = _get_config_value('BINANCE_API_KEY', 'your_api_key_here')
new_api_secret = _get_config_value('BINANCE_API_SECRET', 'your_api_secret_here')
# 如果获取到有效值,更新全局变量
if new_api_key and new_api_key != 'your_api_key_here':
BINANCE_API_KEY = new_api_key
if new_api_secret and new_api_secret != 'your_api_secret_here':
BINANCE_API_SECRET = new_api_secret
USE_TESTNET = _get_config_value('USE_TESTNET', False) if _get_config_value('USE_TESTNET') is not None else os.getenv('USE_TESTNET', 'False').lower() == 'true'
TRADING_CONFIG = _get_trading_config()
# 重新加载 Redis 配置
REDIS_URL = _get_config_value('REDIS_URL', os.getenv('REDIS_URL', REDIS_URL))
REDIS_USE_TLS = _get_config_value('REDIS_USE_TLS', False) if _get_config_value('REDIS_USE_TLS') is not None else os.getenv('REDIS_USE_TLS', 'False').lower() == 'true'
REDIS_SSL_CERT_REQS = _get_config_value('REDIS_SSL_CERT_REQS', REDIS_SSL_CERT_REQS)
REDIS_SSL_CA_CERTS = _get_config_value('REDIS_SSL_CA_CERTS', REDIS_SSL_CA_CERTS)
REDIS_USERNAME = _get_config_value('REDIS_USERNAME', os.getenv('REDIS_USERNAME', REDIS_USERNAME))
REDIS_PASSWORD = _get_config_value('REDIS_PASSWORD', os.getenv('REDIS_PASSWORD', REDIS_PASSWORD))
REDIS_SELECT = _get_config_value('REDIS_SELECT', os.getenv('REDIS_SELECT', REDIS_SELECT))
# 确保默认值
for key, value in defaults.items():
if key not in TRADING_CONFIG:
TRADING_CONFIG[key] = value
# 连接配置
CONNECTION_TIMEOUT = int(os.getenv('CONNECTION_TIMEOUT', '30')) # 连接超时时间(秒)
CONNECTION_RETRIES = int(os.getenv('CONNECTION_RETRIES', '3')) # 连接重试次数
# Redis 缓存配置(优先从数据库,回退到环境变量和默认值)
REDIS_URL = _get_config_value('REDIS_URL', os.getenv('REDIS_URL', 'redis://localhost:6379'))
REDIS_USE_TLS = _get_config_value('REDIS_USE_TLS', False) if _get_config_value('REDIS_USE_TLS') is not None else os.getenv('REDIS_USE_TLS', 'False').lower() == 'true'
REDIS_SSL_CERT_REQS = _get_config_value('REDIS_SSL_CERT_REQS', 'required')
REDIS_SSL_CA_CERTS = _get_config_value('REDIS_SSL_CA_CERTS', None)
REDIS_USERNAME = _get_config_value('REDIS_USERNAME', os.getenv('REDIS_USERNAME', None))
REDIS_PASSWORD = _get_config_value('REDIS_PASSWORD', os.getenv('REDIS_PASSWORD', None))
REDIS_SELECT = _get_config_value('REDIS_SELECT', os.getenv('REDIS_SELECT', 0))
# 日志配置
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
LOG_FILE = 'trading_bot.log'