382 lines
15 KiB
Python
382 lines
15 KiB
Python
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
|
|
from .common import logger
|
|
|
|
|
|
def get_env_or_none(key: str) -> str | None:
|
|
"""환경변수를 로드하되, 없으면 None 반환 (빈 문자열도 None 처리)"""
|
|
value = os.getenv(key)
|
|
return value if value else None
|
|
|
|
|
|
def get_default_config() -> dict:
|
|
"""기본 설정 반환 (config.json 로드 실패 시 사용)"""
|
|
return {
|
|
"symbols_file": "config/symbols.txt",
|
|
"symbol_delay": 1.0,
|
|
"candle_count": 200,
|
|
"buy_check_interval_minutes": 240,
|
|
"stop_loss_check_interval_minutes": 60,
|
|
"profit_taking_check_interval_minutes": 240,
|
|
"balance_warning_interval_hours": 24,
|
|
"min_amount_threshold": 1e-8,
|
|
"loop": True,
|
|
"dry_run": True,
|
|
"max_threads": 3,
|
|
"telegram_parse_mode": "HTML",
|
|
"trading_mode": "signal_only",
|
|
"auto_trade": {
|
|
"enabled": False,
|
|
"buy_enabled": False,
|
|
"buy_amount_krw": 10000,
|
|
"min_order_value_krw": 5000,
|
|
"telegram_max_retries": 3,
|
|
"order_monitor_max_errors": 5,
|
|
},
|
|
}
|
|
|
|
|
|
def validate_config(cfg: dict) -> tuple[bool, str]:
|
|
"""설정 파일의 필수 항목을 검증합니다 (MEDIUM-001 + HIGH-002)
|
|
|
|
Args:
|
|
cfg: 설정 딕셔너리
|
|
|
|
Returns:
|
|
(is_valid, error_message)
|
|
- is_valid: 검증 통과 여부
|
|
- error_message: 오류 메시지 (성공 시 빈 문자열)
|
|
"""
|
|
required_keys = [
|
|
"buy_check_interval_minutes",
|
|
"stop_loss_check_interval_minutes",
|
|
"profit_taking_check_interval_minutes",
|
|
"dry_run",
|
|
"auto_trade",
|
|
]
|
|
|
|
# 필수 항목 확인
|
|
for key in required_keys:
|
|
if key not in cfg:
|
|
return False, f"필수 설정 항목 누락: '{key}'"
|
|
|
|
# 범위 검증
|
|
try:
|
|
buy_interval = cfg.get("buy_check_interval_minutes", 0)
|
|
if not isinstance(buy_interval, (int, float)) or buy_interval < 1:
|
|
return False, "buy_check_interval_minutes는 1 이상이어야 합니다"
|
|
|
|
stop_loss_interval = cfg.get("stop_loss_check_interval_minutes", 0)
|
|
if not isinstance(stop_loss_interval, (int, float)) or stop_loss_interval < 1:
|
|
return False, "stop_loss_check_interval_minutes는 1 이상이어야 합니다"
|
|
|
|
profit_interval = cfg.get("profit_taking_check_interval_minutes", 0)
|
|
if not isinstance(profit_interval, (int, float)) or profit_interval < 1:
|
|
return False, "profit_taking_check_interval_minutes는 1 이상이어야 합니다"
|
|
|
|
# auto_trade 설정 검증
|
|
auto_trade = cfg.get("auto_trade", {})
|
|
if not isinstance(auto_trade, dict):
|
|
return False, "auto_trade는 딕셔너리 형식이어야 합니다"
|
|
|
|
# confirm 설정 검증
|
|
confirm = cfg.get("confirm", {})
|
|
if isinstance(confirm, dict):
|
|
if not isinstance(confirm.get("confirm_stop_loss", False), bool):
|
|
return False, "confirm_stop_loss는 boolean 타입이어야 합니다"
|
|
else:
|
|
return False, "confirm 설정은 딕셔너리 형식이어야 합니다"
|
|
|
|
# dry_run 타입 검증
|
|
if not isinstance(cfg.get("dry_run"), bool):
|
|
return False, "dry_run은 true 또는 false여야 합니다"
|
|
|
|
# ============================================================================
|
|
# HIGH-002: 추가 검증 로직 (상호 의존성, 논리적 모순, 위험 설정)
|
|
# ============================================================================
|
|
|
|
# 1. Auto Trade 활성화 시 API 키 필수 검증
|
|
if auto_trade.get("enabled") or auto_trade.get("buy_enabled"):
|
|
access_key = get_env_or_none("UPBIT_ACCESS_KEY")
|
|
secret_key = get_env_or_none("UPBIT_SECRET_KEY")
|
|
if not access_key or not secret_key:
|
|
return False, "auto_trade 활성화 시 UPBIT_ACCESS_KEY와 UPBIT_SECRET_KEY 환경변수 필수"
|
|
|
|
# 2. 손절/익절 주기 논리 검증 (손절은 더 자주 체크해야 안전)
|
|
if stop_loss_interval > profit_interval:
|
|
logger.warning(
|
|
"[설정 경고] 손절 주기(%d분)가 익절 주기(%d분)보다 깁니다. "
|
|
"급락 시 손절이 늦어질 수 있으므로 손절을 더 자주 체크하는 것이 안전합니다.",
|
|
stop_loss_interval,
|
|
profit_interval,
|
|
)
|
|
|
|
# 3. 스레드 수 검증 (과도한 스레드는 Rate Limit 초과 위험)
|
|
max_threads = cfg.get("max_threads", 3)
|
|
if not isinstance(max_threads, int) or max_threads < 1:
|
|
return False, "max_threads는 1 이상의 정수여야 합니다"
|
|
|
|
if max_threads > 10:
|
|
logger.warning(
|
|
"[설정 경고] max_threads=%d는 과도할 수 있습니다. "
|
|
"Upbit API Rate Limit(초당 8회, 분당 590회)을 고려하면 10 이하 권장.",
|
|
max_threads,
|
|
)
|
|
|
|
# 4. 최소 주문 금액 검증
|
|
min_order = auto_trade.get("min_order_value_krw")
|
|
if min_order is not None:
|
|
if not isinstance(min_order, (int, float)) or min_order < 5000:
|
|
return False, "min_order_value_krw는 5000원 이상이어야 합니다 (Upbit 최소 주문 금액)"
|
|
|
|
# 5. 매수 금액 검증
|
|
buy_amount = auto_trade.get("buy_amount_krw")
|
|
if buy_amount is not None:
|
|
if not isinstance(buy_amount, (int, float)) or buy_amount < 5000:
|
|
return False, "buy_amount_krw는 5000원 이상이어야 합니다"
|
|
|
|
# 최소 주문 금액보다 매수 금액이 작은 경우
|
|
if min_order and buy_amount < min_order:
|
|
logger.warning(
|
|
"[설정 경고] buy_amount_krw(%d원)가 min_order_value_krw(%d원)보다 작습니다. "
|
|
"주문이 실행되지 않을 수 있습니다.",
|
|
buy_amount,
|
|
min_order,
|
|
)
|
|
|
|
except (TypeError, ValueError) as e:
|
|
return False, f"설정값 타입 오류: {e}"
|
|
|
|
return True, ""
|
|
|
|
|
|
def load_config() -> dict:
|
|
paths = [os.path.join("config", "config.json"), "config.json"]
|
|
example_paths = [os.path.join("config", "config.example.json"), "config.example.json"]
|
|
for p in paths:
|
|
if os.path.exists(p):
|
|
try:
|
|
with open(p, encoding="utf-8") as f:
|
|
cfg = json.load(f)
|
|
|
|
# ✅ MEDIUM-001: 설정 파일 검증
|
|
is_valid, error_msg = validate_config(cfg)
|
|
if not is_valid:
|
|
logger.error("설정 파일 검증 실패: %s. 기본 설정 사용.", error_msg)
|
|
return get_default_config()
|
|
|
|
logger.info("설정 파일 로드 및 검증 완료: %s", p)
|
|
return cfg
|
|
except json.JSONDecodeError as e:
|
|
logger.error("설정 파일 JSON 파싱 실패: %s, 기본 설정 사용", e)
|
|
return get_default_config()
|
|
for p in example_paths:
|
|
if os.path.exists(p):
|
|
try:
|
|
with open(p, encoding="utf-8") as f:
|
|
cfg = json.load(f)
|
|
logger.warning("기본 설정 없음; 예제 사용: %s", p)
|
|
return cfg
|
|
except json.JSONDecodeError:
|
|
pass
|
|
logger.warning("설정 파일 없음: config/config.json, 기본 설정 사용")
|
|
return get_default_config()
|
|
|
|
|
|
def read_symbols(path: str) -> list:
|
|
syms = []
|
|
syms_set = set()
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
for line in f:
|
|
s = line.strip()
|
|
if not s or s.startswith("#"):
|
|
continue
|
|
if s in syms_set:
|
|
logger.warning("[SYSTEM] 중복 심볼 무시: %s", s)
|
|
continue
|
|
syms_set.add(s)
|
|
syms.append(s)
|
|
logger.info("[SYSTEM] 심볼 %d개 로드: %s", len(syms), path)
|
|
except Exception as e:
|
|
logger.exception("[ERROR] 심볼 로드 실패: %s", e)
|
|
return syms
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeConfig:
|
|
timeframe: str
|
|
indicator_timeframe: str
|
|
candle_count: int
|
|
symbol_delay: float
|
|
interval: int
|
|
loop: bool
|
|
dry_run: bool
|
|
max_threads: int
|
|
telegram_parse_mode: str | None
|
|
trading_mode: str
|
|
telegram_bot_token: str | None
|
|
telegram_chat_id: str | None
|
|
upbit_access_key: str | None
|
|
upbit_secret_key: str | None
|
|
aggregate_alerts: bool = False
|
|
benchmark: bool = False
|
|
telegram_test: bool = False
|
|
config: dict = None # 원본 config 포함
|
|
|
|
|
|
def validate_telegram_token(token: str) -> bool:
|
|
"""텔레그램 봇 토큰 형식 검증 (형식: 숫자:영숫자_-)"""
|
|
import re
|
|
|
|
if not token:
|
|
return False
|
|
# 텔레그램 토큰 형식: 123456789:ABCdefGHIjklMNOpqrsTUVwxyz-_1234567
|
|
# 첫 부분: 8-10자리 숫자, 두 번째 부분: 35자 이상
|
|
pattern = r"^\d{8,10}:[A-Za-z0-9_-]{35,}$"
|
|
return bool(re.match(pattern, token))
|
|
|
|
|
|
def build_runtime_config(cfg_dict: dict) -> RuntimeConfig:
|
|
# 설정값 검증
|
|
candle_count = int(cfg_dict.get("candle_count", 200))
|
|
if candle_count < 1:
|
|
logger.warning("[WARNING] candle_count는 1 이상이어야 합니다. 기본값 200 사용")
|
|
candle_count = 200
|
|
|
|
max_threads = int(cfg_dict.get("max_threads", 3))
|
|
if max_threads < 1:
|
|
logger.warning("[WARNING] max_threads는 1 이상이어야 합니다. 기본값 3 사용")
|
|
max_threads = 3
|
|
|
|
symbol_delay = float(cfg_dict.get("symbol_delay", 1.0))
|
|
if symbol_delay < 0:
|
|
logger.warning("[WARNING] symbol_delay는 0 이상이어야 합니다. 기본값 1.0 사용")
|
|
symbol_delay = 1.0
|
|
|
|
# timeframe은 동적으로 결정되므로 기본값만 설정 (실제로는 매수/매도 주기에 따라 변경됨)
|
|
timeframe = "1h" # 기본값
|
|
aggregate_alerts = bool(cfg_dict.get("aggregate_alerts", False)) or bool(
|
|
os.getenv("AGGREGATE_ALERTS", "False").lower() in ("1", "true", "yes")
|
|
)
|
|
benchmark = bool(cfg_dict.get("benchmark", False))
|
|
telegram_test = os.getenv("TELEGRAM_TEST", "0") == "1"
|
|
|
|
# auto_trade 서브 설정 논리 관계 검증 및 교정
|
|
at = cfg_dict.get("auto_trade", {}) or {}
|
|
loss_threshold = float(at.get("loss_threshold", -5.0))
|
|
p1 = float(at.get("profit_threshold_1", 10.0))
|
|
p2 = float(at.get("profit_threshold_2", 30.0))
|
|
d1 = float(at.get("drawdown_1", 5.0))
|
|
d2 = float(at.get("drawdown_2", 15.0))
|
|
|
|
# 손절 임계값 검증 (음수여야 함)
|
|
if loss_threshold >= 0:
|
|
logger.warning("[WARNING] loss_threshold(%.2f)는 음수여야 합니다 (예: -5.0). 기본값 -5.0 적용", loss_threshold)
|
|
loss_threshold = -5.0
|
|
elif loss_threshold < -50:
|
|
logger.warning(
|
|
"[WARNING] loss_threshold(%.2f)가 너무 작습니다 (최대 손실 50%% 초과). 극단적인 손절선입니다.",
|
|
loss_threshold,
|
|
)
|
|
|
|
# 수익률 임계값 검증 (양수, 순서 관계, 합리적 범위)
|
|
if p1 <= 0 or p2 <= 0:
|
|
logger.warning("[WARNING] 수익률 임계값은 양수여야 합니다 -> 기본값 10/30 재설정")
|
|
p1, p2 = 10.0, 30.0
|
|
elif p1 >= p2:
|
|
logger.warning(
|
|
"[WARNING] profit_threshold_1(%.2f) < profit_threshold_2(%.2f) 조건 위반 -> 기본값 10/30 적용", p1, p2
|
|
)
|
|
p1, p2 = 10.0, 30.0
|
|
elif p1 < 5 or p2 > 200:
|
|
logger.warning(
|
|
"[WARNING] 수익률 임계값 범위 권장 벗어남 (p1=%.2f, p2=%.2f). 권장 범위: 5%% <= p1 < p2 <= 200%%", p1, p2
|
|
)
|
|
|
|
# 드로우다운 임계값 검증 (양수, 순서 관계, 합리적 범위)
|
|
if d1 <= 0 or d2 <= 0:
|
|
logger.warning("[WARNING] drawdown 임계값은 양수여야 합니다 -> 기본값 5/15 재설정")
|
|
d1, d2 = 5.0, 15.0
|
|
elif d1 >= d2:
|
|
logger.warning("[WARNING] drawdown_1(%.2f) < drawdown_2(%.2f) 조건 위반 -> 기본값 5/15 적용", d1, d2)
|
|
d1, d2 = 5.0, 15.0
|
|
elif d1 > 20 or d2 > 50:
|
|
logger.warning(
|
|
"[WARNING] drawdown 값이 너무 큽니다 (d1=%.2f, d2=%.2f). "
|
|
"최대 손실 허용치를 확인하세요. 권장: d1 <= 20%%, d2 <= 50%%",
|
|
d1,
|
|
d2,
|
|
)
|
|
|
|
# 교정된 값 반영
|
|
at["profit_threshold_1"] = p1
|
|
at["profit_threshold_2"] = p2
|
|
at["drawdown_1"] = d1
|
|
at["drawdown_2"] = d2
|
|
cfg_dict["auto_trade"] = at
|
|
|
|
# 환경변수 로드
|
|
telegram_token = os.getenv("TELEGRAM_BOT_TOKEN")
|
|
telegram_chat = os.getenv("TELEGRAM_CHAT_ID")
|
|
upbit_access = os.getenv("UPBIT_ACCESS_KEY")
|
|
upbit_secret = os.getenv("UPBIT_SECRET_KEY")
|
|
|
|
# dry_run 및 trading_mode 확인
|
|
dry_run = bool(cfg_dict.get("dry_run", False))
|
|
trading_mode = cfg_dict.get("trading_mode", "signal_only")
|
|
|
|
# 실거래 모드 시 필수 API 키 검증
|
|
if not dry_run and trading_mode in ("auto_trade", "mixed"):
|
|
if not (upbit_access and upbit_secret):
|
|
raise ValueError(
|
|
"[CRITICAL] 실거래 모드(dry_run=False)에서는 UPBIT_ACCESS_KEY 및 "
|
|
"UPBIT_SECRET_KEY 환경변수가 필수입니다. "
|
|
".env 파일을 확인하거나 환경변수를 설정하십시오."
|
|
)
|
|
logger.info(
|
|
"[SECURITY] Upbit API 키 로드 완료 (access_key 길이: %d, secret_key 길이: %d)",
|
|
len(upbit_access),
|
|
len(upbit_secret),
|
|
)
|
|
|
|
# 텔레그램 토큰 검증 (설정되어 있으면)
|
|
if telegram_token and not validate_telegram_token(telegram_token):
|
|
logger.warning(
|
|
"[WARNING] TELEGRAM_BOT_TOKEN 형식이 올바르지 않습니다 (형식: 숫자:영숫자_-). "
|
|
"알림 전송이 실패할 수 있습니다."
|
|
)
|
|
|
|
return RuntimeConfig(
|
|
timeframe=timeframe,
|
|
indicator_timeframe=timeframe,
|
|
candle_count=candle_count,
|
|
symbol_delay=symbol_delay,
|
|
interval=int(cfg_dict.get("interval", 60)),
|
|
loop=bool(cfg_dict.get("loop", False)),
|
|
dry_run=dry_run,
|
|
max_threads=max_threads,
|
|
telegram_parse_mode=cfg_dict.get("telegram_parse_mode"),
|
|
trading_mode=trading_mode,
|
|
telegram_bot_token=telegram_token,
|
|
telegram_chat_id=telegram_chat,
|
|
upbit_access_key=upbit_access,
|
|
upbit_secret_key=upbit_secret,
|
|
aggregate_alerts=aggregate_alerts,
|
|
benchmark=benchmark,
|
|
telegram_test=telegram_test,
|
|
config=cfg_dict,
|
|
)
|
|
|
|
|
|
def get_symbols_file(config: dict) -> str:
|
|
"""Determine the symbols file path with fallback logic."""
|
|
default_path = "config/symbols.txt" if os.path.exists("config/symbols.txt") else "symbols.txt"
|
|
return config.get("symbols_file", default_path)
|
|
|
|
|
|
# Define valid trading modes as constants
|
|
TRADING_MODES = ("signal_only", "auto_trade", "mixed")
|