36a3e2b4a1
Changes: - Added a new API endpoint for managing permanent subscriptions, allowing users to enable or disable subscriptions dynamically. - Implemented a function to fill candle data from Kiwoom, ensuring that only relevant data is inserted into the database. - Introduced a mechanism to handle master subscription states, improving the management of subscription statuses. - Updated the database schema to include new fields for managing subscription states and order book filtering. Impact: - These enhancements improve the flexibility and reliability of the trading system, allowing for better management of subscriptions and order book data, while reducing the risk of data inconsistencies. 히스토리 align 제거 븅신같은 초기설계 아예 제거 진입모드에 구멍메움 호가진입을 켜도 호가가 안들어올때 호가 안보고 그냥 사버림
316 lines
10 KiB
Python
316 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Optuna 결과(mode_combo top_pnls / mode·best PnL) → 당일손익 다단트레일 추천.
|
|
|
|
- 탐색 축에는 넣지 않음 (그리드/TPE 제외).
|
|
- JSON·Optuna 탭 미리보기용 추천만 생성.
|
|
- DB 반영은 apply-best / 웹「DB 적용」시에만 ({SID}_DAILY_PROFIT_* , 마스터 키 미사용).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import statistics
|
|
from typing import Any, Dict, List, Optional, Sequence
|
|
|
|
logger = logging.getLogger("optuna_daily_trail_recommend")
|
|
|
|
# Optuna strategy 슬러그 → DB 접두
|
|
_STRATEGY_PREFIX: Dict[str, str] = {
|
|
"tail": "SHORT",
|
|
"short": "SHORT",
|
|
"scalp": "SCALP",
|
|
"scalping": "SCALP",
|
|
"momentum": "MOMENTUM",
|
|
"us_momentum": "US_MOMENTUM",
|
|
"breakout": "BREAKOUT",
|
|
}
|
|
|
|
|
|
def strategy_daily_profit_prefix(strategy: str) -> str:
|
|
s = str(strategy or "").strip().lower()
|
|
return _STRATEGY_PREFIX.get(s, s.upper() if s else "SHORT")
|
|
|
|
|
|
def _env_float(key: str, default: float) -> float:
|
|
try:
|
|
from kis_trader.utils.env import get_env_float
|
|
return float(get_env_float(key, float(default)))
|
|
except Exception:
|
|
return float(default)
|
|
|
|
|
|
def _env_int(key: str, default: int) -> int:
|
|
try:
|
|
from kis_trader.utils.env import get_env_int
|
|
return int(get_env_int(key, int(default)))
|
|
except Exception:
|
|
return int(default)
|
|
|
|
|
|
def _env_str(key: str, default: str) -> str:
|
|
try:
|
|
from kis_trader.utils.env import get_env_from_db
|
|
raw = get_env_from_db(key, default)
|
|
if raw is None or str(raw).strip() in ("", "None"):
|
|
return str(default)
|
|
return str(raw).strip()
|
|
except Exception:
|
|
return str(default)
|
|
|
|
|
|
def _env_bool(key: str, default: bool = True) -> bool:
|
|
try:
|
|
from kis_trader.utils.env import get_env_bool
|
|
return bool(get_env_bool(key, bool(default)))
|
|
except Exception:
|
|
return bool(default)
|
|
|
|
|
|
def _round_arm_krw(value: float, step: int) -> int:
|
|
step = max(1000, int(step or 5000))
|
|
if value <= 0:
|
|
return 0
|
|
return int(max(step, round(float(value) / step) * step))
|
|
|
|
|
|
def _parse_drops(raw: str) -> List[float]:
|
|
parts = [p.strip() for p in str(raw or "").split(",") if p.strip()]
|
|
out: List[float] = []
|
|
for p in parts[:3]:
|
|
try:
|
|
out.append(max(5.0, min(80.0, float(p))))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
while len(out) < 3:
|
|
out.append([40.0, 30.0, 20.0][len(out)])
|
|
return out[:3]
|
|
|
|
|
|
def _positive_pnls(vals: Sequence[Any]) -> List[float]:
|
|
out: List[float] = []
|
|
for v in vals or []:
|
|
try:
|
|
x = float(v)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if x > 0 and abs(x) < 1e15:
|
|
out.append(x)
|
|
return out
|
|
|
|
|
|
def recommend_daily_trail_tiers(
|
|
*,
|
|
top_pnls: Optional[Sequence[Any]] = None,
|
|
mode_pnl: Optional[float] = None,
|
|
best_pnl: Optional[float] = None,
|
|
strategy: str = "",
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
추천 공식:
|
|
anchor = max(mode_pnl, median(top_pnls), best_pnl * BEST_FRAC)
|
|
arm = round(anchor * ARM_FRAC) (step·min_arm 적용)
|
|
tiers = arm:d1, (arm*2):d2, (arm*4):d3
|
|
PnL 전부 ≤0 이면 ok=False (적용 스킵 대상).
|
|
"""
|
|
tops = _positive_pnls(list(top_pnls or []))
|
|
med = float(statistics.median(tops)) if tops else 0.0
|
|
mode_v = float(mode_pnl or 0.0)
|
|
best_v = float(best_pnl or 0.0)
|
|
best_frac = _env_float("OPTUNA_DAILY_TRAIL_BEST_FRAC", 0.70)
|
|
arm_frac = _env_float("OPTUNA_DAILY_TRAIL_ARM_FRAC", 0.60)
|
|
step = _env_int("OPTUNA_DAILY_TRAIL_ARM_STEP", 5000)
|
|
min_arm = _env_int("OPTUNA_DAILY_TRAIL_MIN_ARM", 10000)
|
|
drops = _parse_drops(_env_str("OPTUNA_DAILY_TRAIL_TIER_DROPS", "40,30,20"))
|
|
|
|
candidates = [x for x in (mode_v, med, best_v * best_frac) if x > 0]
|
|
anchor = max(candidates) if candidates else 0.0
|
|
arm_raw = anchor * arm_frac if anchor > 0 else 0.0
|
|
arm = _round_arm_krw(arm_raw, step)
|
|
if arm > 0:
|
|
arm = max(arm, min_arm)
|
|
arm = _round_arm_krw(float(arm), step)
|
|
|
|
prefix = strategy_daily_profit_prefix(strategy)
|
|
if arm <= 0:
|
|
return {
|
|
"ok": False,
|
|
"strategy": str(strategy or "").strip().lower(),
|
|
"prefix": prefix,
|
|
"reason": "양수 PnL 앵커 없음 — 다단트레일 추천 생략",
|
|
"anchor_krw": 0,
|
|
"arm_krw": 0,
|
|
"tiers": "",
|
|
"mode": "trailing",
|
|
"enabled": False,
|
|
"inputs": {
|
|
"mode_pnl": mode_v,
|
|
"best_pnl": best_v,
|
|
"top_median": med,
|
|
"top_pnls_head": tops[:10],
|
|
},
|
|
"formula": {
|
|
"arm_frac": arm_frac,
|
|
"best_frac": best_frac,
|
|
"step": step,
|
|
"min_arm": min_arm,
|
|
"drops": drops,
|
|
},
|
|
"note": "apply 시에도 TRAIL 미기록 (수익 앵커 없음)",
|
|
}
|
|
|
|
t1, t2, t3 = int(arm), int(arm * 2), int(arm * 4)
|
|
d1, d2, d3 = drops
|
|
tiers = f"{t1}:{d1:.0f},{t2}:{d2:.0f},{t3}:{d3:.0f}"
|
|
return {
|
|
"ok": True,
|
|
"strategy": str(strategy or "").strip().lower(),
|
|
"prefix": prefix,
|
|
"reason": "",
|
|
"anchor_krw": int(anchor),
|
|
"arm_krw": int(arm),
|
|
"tiers": tiers,
|
|
"mode": "trailing",
|
|
"enabled": True,
|
|
"inputs": {
|
|
"mode_pnl": mode_v,
|
|
"best_pnl": best_v,
|
|
"top_median": med,
|
|
"top_pnls_head": tops[:10],
|
|
},
|
|
"formula": {
|
|
"arm_frac": arm_frac,
|
|
"best_frac": best_frac,
|
|
"step": step,
|
|
"min_arm": min_arm,
|
|
"drops": drops,
|
|
},
|
|
"note": (
|
|
f"apply 시 {prefix}_DAILY_PROFIT_TRAIL_TIERS={tiers} "
|
|
f"(ENABLED=true, MODE=trailing). 운영 UI에서 끄거나 수정 가능."
|
|
),
|
|
}
|
|
|
|
|
|
def recommend_from_optuna_out_data(out_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Optuna 결과 dict(mode_combo·results)에서 추천 생성."""
|
|
data = out_data or {}
|
|
strat = str(data.get("strategy") or "").strip().lower()
|
|
mc = data.get("mode_combo") or {}
|
|
top_pnls = list(mc.get("top_pnls") or [])
|
|
bt = mc.get("backtest") or {}
|
|
mode_pnl = bt.get("total_pnl")
|
|
vs = mc.get("vs_best") or {}
|
|
best_pnl = vs.get("best_pnl")
|
|
if best_pnl is None:
|
|
res0 = (data.get("results") or [None])[0]
|
|
if res0:
|
|
best_pnl = res0.get("total_pnl")
|
|
if not top_pnls:
|
|
# gated/학습 Top 에서도 보조
|
|
for row in (data.get("results_gated") or data.get("results") or [])[:20]:
|
|
try:
|
|
top_pnls.append(float(row.get("total_pnl") or 0))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return recommend_daily_trail_tiers(
|
|
top_pnls=top_pnls,
|
|
mode_pnl=float(mode_pnl) if mode_pnl is not None else None,
|
|
best_pnl=float(best_pnl) if best_pnl is not None else None,
|
|
strategy=strat,
|
|
)
|
|
|
|
|
|
def attach_daily_trail_recommend(
|
|
out_data: Dict[str, Any],
|
|
*,
|
|
log: Optional[logging.Logger] = None,
|
|
) -> Dict[str, Any]:
|
|
"""out_data 에 daily_trail_recommend 기록 (+ mode_combo 안에도 복사)."""
|
|
lg = log or logger
|
|
rec = recommend_from_optuna_out_data(out_data)
|
|
out_data["daily_trail_recommend"] = rec
|
|
mc = out_data.get("mode_combo")
|
|
if isinstance(mc, dict):
|
|
mc["daily_trail_recommend"] = rec
|
|
if rec.get("ok"):
|
|
lg.info(
|
|
"📅 [다단트레일 추천] %s arm=%s tiers=%s (anchor=%s)",
|
|
rec.get("prefix"),
|
|
rec.get("arm_krw"),
|
|
rec.get("tiers"),
|
|
rec.get("anchor_krw"),
|
|
)
|
|
else:
|
|
lg.info("📅 [다단트레일 추천] 생략 — %s", rec.get("reason") or "n/a")
|
|
return out_data
|
|
|
|
|
|
def build_daily_trail_env_patch(rec: Dict[str, Any]) -> Dict[str, str]:
|
|
"""전략별 DAILY_PROFIT 패치 (마스터 키 없음)."""
|
|
if not rec or not rec.get("ok"):
|
|
return {}
|
|
prefix = str(rec.get("prefix") or "").strip().upper()
|
|
tiers = str(rec.get("tiers") or "").strip()
|
|
if not prefix or not tiers:
|
|
return {}
|
|
mode = str(rec.get("mode") or "trailing").strip().lower() or "trailing"
|
|
return {
|
|
f"{prefix}_DAILY_PROFIT_TARGET_ENABLED": "true",
|
|
f"{prefix}_DAILY_PROFIT_MODE": mode,
|
|
f"{prefix}_DAILY_PROFIT_TRAIL_TIERS": tiers,
|
|
f"{prefix}_DAILY_PROFIT_TRAIL_ARM_KRW": str(int(rec.get("arm_krw") or 0)),
|
|
}
|
|
|
|
|
|
def apply_daily_trail_recommend_patch(
|
|
rec: Dict[str, Any],
|
|
*,
|
|
log: Optional[logging.Logger] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
추천 → DB apply_env_patch.
|
|
OPTUNA_DAILY_TRAIL_APPLY_ON_BEST=false 이면 스킵.
|
|
"""
|
|
lg = log or logger
|
|
if not _env_bool("OPTUNA_DAILY_TRAIL_APPLY_ON_BEST", True):
|
|
return {"applied": False, "reason": "OPTUNA_DAILY_TRAIL_APPLY_ON_BEST=false"}
|
|
patch = build_daily_trail_env_patch(rec)
|
|
if not patch:
|
|
return {"applied": False, "reason": rec.get("reason") or "추천 없음", "recommend": rec}
|
|
try:
|
|
from kis_trader.backtest.param_search_apply_snapshot import apply_env_patch
|
|
env_id = apply_env_patch(patch)
|
|
except Exception as exc:
|
|
lg.warning("⚠️ 다단트레일 추천 DB 반영 실패: %s", exc)
|
|
return {"applied": False, "error": str(exc), "patch": patch, "recommend": rec}
|
|
lg.info(
|
|
"🚀 [Optuna apply] 다단트레일 추천 반영 %s → %s",
|
|
rec.get("prefix"), patch,
|
|
)
|
|
return {"applied": True, "env_id": env_id, "patch": patch, "recommend": rec}
|
|
|
|
|
|
def apply_daily_trail_recommend_from_optuna_json(
|
|
result_json: Optional[str],
|
|
*,
|
|
strategy: str = "",
|
|
log: Optional[logging.Logger] = None,
|
|
) -> Dict[str, Any]:
|
|
"""결과 JSON 경로에서 추천 읽어(또는 재계산) DB 반영."""
|
|
import json
|
|
from pathlib import Path
|
|
|
|
lg = log or logger
|
|
if not result_json or not Path(result_json).is_file():
|
|
return {"applied": False, "reason": "result_json 없음"}
|
|
try:
|
|
data = json.loads(Path(result_json).read_text(encoding="utf-8"))
|
|
except Exception as exc:
|
|
return {"applied": False, "error": str(exc)}
|
|
if strategy and not data.get("strategy"):
|
|
data["strategy"] = strategy
|
|
rec = data.get("daily_trail_recommend")
|
|
if not isinstance(rec, dict) or not rec.get("ok"):
|
|
rec = recommend_from_optuna_out_data(data)
|
|
return apply_daily_trail_recommend_patch(rec, log=lg)
|