166 lines
4.5 KiB
Python
166 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
def _load_dotenv(path: Path) -> None:
|
|
"""간단한 .env 파서 (python-dotenv 의존성 대신 내장 구현)."""
|
|
if not path.exists():
|
|
return
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
key = key.strip()
|
|
value = value.strip().strip('"').strip("'")
|
|
if key and key not in os.environ:
|
|
os.environ[key] = value
|
|
|
|
|
|
_load_dotenv(Path(".env"))
|
|
|
|
|
|
def _env(name: str, default: Any = None) -> Any:
|
|
return os.environ.get(name, default)
|
|
|
|
|
|
class KISConfig:
|
|
app_key: str = ""
|
|
app_secret: str = ""
|
|
account_no: str = ""
|
|
account_code: str = "01"
|
|
hts_id: str = ""
|
|
server_mode: str = "vps"
|
|
|
|
def apply_env(self) -> None:
|
|
app_key = _env("KIS_APP_KEY")
|
|
app_secret = _env("KIS_APP_SECRET")
|
|
account_no = _env("KIS_ACCOUNT_NO")
|
|
account_code = _env("KIS_ACCOUNT_CODE")
|
|
hts_id = _env("KIS_HTS_ID")
|
|
server_mode = _env("KIS_SERVER_MODE")
|
|
|
|
if app_key:
|
|
self.app_key = app_key
|
|
if app_secret:
|
|
self.app_secret = app_secret
|
|
if account_no:
|
|
self.account_no = account_no
|
|
if account_code:
|
|
self.account_code = account_code
|
|
if hts_id:
|
|
self.hts_id = hts_id
|
|
if server_mode:
|
|
self.server_mode = server_mode
|
|
|
|
|
|
class AppConfig:
|
|
name: str = "StockAutomation"
|
|
version: str = "0.1.0"
|
|
host: str = "0.0.0.0"
|
|
port: int = 8000
|
|
debug: bool = False
|
|
db_path: str = "./data/stock.db"
|
|
log_level: str = "INFO"
|
|
|
|
|
|
class CollectorConfig:
|
|
interval_seconds: int = 5
|
|
market_open_hour: int = 9
|
|
market_close_hour: int = 15
|
|
market_close_minute: int = 30
|
|
|
|
|
|
class StrategyConfig:
|
|
check_interval_seconds: int = 3
|
|
max_daily_trades: int = 50
|
|
default_order_type: str = "00"
|
|
|
|
|
|
class TradingConfig:
|
|
max_order_amount: int = 10_000_000
|
|
min_order_amount: int = 100_000
|
|
slippage_percent: float = 0.1
|
|
|
|
|
|
class RateLimitConfig:
|
|
requests_per_second: float = 1.0
|
|
retry_delay: float = 1.5
|
|
|
|
|
|
class Settings:
|
|
def __init__(self) -> None:
|
|
self.kis = KISConfig()
|
|
self.app = AppConfig()
|
|
self.collector = CollectorConfig()
|
|
self.strategy = StrategyConfig()
|
|
self.trading = TradingConfig()
|
|
self.rate_limit = RateLimitConfig()
|
|
self._load_yaml()
|
|
self.kis.apply_env()
|
|
|
|
def _load_yaml(self) -> None:
|
|
yaml_path = Path("config.yaml")
|
|
if not yaml_path.exists():
|
|
return
|
|
|
|
with open(yaml_path, encoding="utf-8") as f:
|
|
data: dict[str, Any] = yaml.safe_load(f) or {}
|
|
|
|
app = data.get("app", {})
|
|
for key, val in app.items():
|
|
if hasattr(self.app, key):
|
|
setattr(self.app, key, val)
|
|
|
|
kis = data.get("kis", {})
|
|
for key, val in kis.items():
|
|
if key == "rate_limit":
|
|
continue
|
|
if hasattr(self.kis, key):
|
|
setattr(self.kis, key, val)
|
|
|
|
rl = kis.get("rate_limit", {})
|
|
if "requests_per_second" in rl:
|
|
self.rate_limit.requests_per_second = rl["requests_per_second"]
|
|
if "retry_delay" in rl:
|
|
self.rate_limit.retry_delay = rl["retry_delay"]
|
|
|
|
collector = data.get("collector", {})
|
|
for key, val in collector.items():
|
|
if hasattr(self.collector, key):
|
|
setattr(self.collector, key, val)
|
|
|
|
strategy = data.get("strategies", {})
|
|
for key, val in strategy.items():
|
|
if hasattr(self.strategy, key):
|
|
setattr(self.strategy, key, val)
|
|
|
|
trading = data.get("trading", {})
|
|
for key, val in trading.items():
|
|
if hasattr(self.trading, key):
|
|
setattr(self.trading, key, val)
|
|
|
|
logging_cfg = data.get("logging", {})
|
|
if "level" in logging_cfg:
|
|
self.app.log_level = logging_cfg["level"]
|
|
|
|
|
|
settings = Settings()
|
|
|
|
|
|
def get_base_url() -> str:
|
|
if settings.kis.server_mode == "real":
|
|
return "https://openapi.koreainvestment.com:9443"
|
|
return "https://openapivts.koreainvestment.com:29443"
|
|
|
|
|
|
def get_ws_url() -> str:
|
|
if settings.kis.server_mode == "real":
|
|
return "ws://ops.koreainvestment.com:21000"
|
|
return "ws://ops.koreainvestment.com:31000"
|