최초 프로젝트 업로드 (Script Auto Commit)
This commit is contained in:
259
src/config.py
Normal file
259
src/config.py
Normal file
@@ -0,0 +1,259 @@
|
||||
import os, json
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
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 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, "r", encoding="utf-8") as f:
|
||||
cfg = json.load(f)
|
||||
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, "r", 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, "r", 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: Optional[str]
|
||||
trading_mode: str
|
||||
telegram_bot_token: Optional[str]
|
||||
telegram_chat_id: Optional[str]
|
||||
upbit_access_key: Optional[str]
|
||||
upbit_secret_key: Optional[str]
|
||||
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")
|
||||
Reference in New Issue
Block a user