예제 소스

rl_policy_teacher.py

"""
config/policy.md 를 반영한 규칙 기반 teacher 분포(로짓) + 체크포인트 메타데이터.

문서의 서술(수수료·MA 수렴·MA60 방향·에너지 등)을 차트 수치로 근사해
정책 네트워크에 KL/교차엔트로피 보조 손실을 줄 때 사용합니다.
"""

from __future__ import annotations

import hashlib
import json
import logging
import math
from pathlib import Path

import numpy as np
import pandas as pd
import torch
import torch.nn.functional as F

import main_llm_db_test_ollama as ollama_bt

logger = logging.getLogger(__name__)

# action index: 매도=0, 매수=1, 유지=2 (_opinion_to_signal 과 동일)
ACTION_SELL, ACTION_BUY, ACTION_HOLD = 0, 1, 2


def load_policy_markdown(path: str | Path | None) -> str:
    if not path:
        return ""
    p = Path(path)
    if not p.is_file():
        logger.warning("policy 문서 없음: %s", p)
        return ""
    try:
        return p.read_text(encoding="utf-8").strip()
    except OSError as e:
        logger.warning("policy 문서 읽기 실패: %s", e)
        return ""


def policy_document_fingerprint(path: str | Path | None) -> dict:
    if not path:
        return {"path": None, "missing": True}
    p = Path(path)
    if not p.is_file():
        return {"path": str(p), "missing": True}
    raw = p.read_bytes()
    return {
        "path": str(p.resolve()),
        "sha256": hashlib.sha256(raw).hexdigest(),
        "size": len(raw),
        "mtime_ns": p.stat().st_mtime_ns,
        "missing": False,
    }


def write_rl_training_meta(
    checkpoint_path: str | Path,
    *,
    policy_doc_path: str | None,
    extra: dict | None = None,
) -> None:
    """학습에 사용한 policy 문서 지문을 체크포인트 옆 JSON 으로 저장."""
    ckpt = Path(checkpoint_path)
    meta_path = ckpt.with_name(ckpt.name + ".train_meta.json")
    payload = {
        "checkpoint": str(ckpt.resolve()),
        "policy_document": policy_document_fingerprint(policy_doc_path),
    }
    if extra:
        payload["extra"] = extra
    meta_path.parent.mkdir(parents=True, exist_ok=True)
    with open(meta_path, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
    logger.info("RL 학습 메타 저장: %s", meta_path)


def _safe_float(x, default: float = 0.0) -> float:
    try:
        v = float(x)
        if math.isnan(v) or math.isinf(v):
            return default
        return v
    except (TypeError, ValueError):
        return default


def _bool01(x) -> float:
    if x is None or (isinstance(x, float) and (math.isnan(x) or math.isinf(x))):
        return 0.0
    if isinstance(x, (bool, np.bool_)):
        return 1.0 if x else 0.0
    if isinstance(x, (int, np.integer)):
        return 1.0 if x else 0.0
    if isinstance(x, str):
        return 1.0 if x.lower() in ("true", "1", "yes") else 0.0
    return 1.0 if x else 0.0


def opinion_to_action_index(opinion: str | None) -> int:
    if opinion == "매도":
        return ACTION_SELL
    if opinion == "매수":
        return ACTION_BUY
    return ACTION_HOLD


def teacher_logits_from_chart(
    df: pd.DataFrame,
    bar_idx: int,
    portfolio: ollama_bt.PaperLlmPortfolio,
    price: float,
) -> np.ndarray:
    """
    policy.md 규칙을 상태 기반 스칼라 점수로 근사 → 3클래스 로짓.

    - 수수료(약 1% 왕복): 미실현 +0~1% 구간에서 성급한 매도 억제
    - MA20/60/120 수렴·발산: '에너지' 근사 → 수렴 시 관망·매수 절제, 발산 시 변동성 주의
    - 고점 부근 수렴: 매수 억제
    - MA60 기울기: 하향 시 매수 감점·매도 가점, 상향 시 상승 국면 가점
    - 극단 라벨(ext_crash_signal 등): 하락 위험 시 매도 가점
    """
    row = df.iloc[bar_idx]
    close = _safe_float(row.get("close", price), price)
    open_ = _safe_float(row.get("open", close), close)
    high = _safe_float(row.get("high", close), close)
    low = _safe_float(row.get("low", close), close)
    ma20 = _safe_float(row.get("ma20", close), close)
    ma60 = _safe_float(row.get("ma60", close), close)
    ma120 = _safe_float(row.get("ma120", ma60), ma60)

    j = bar_idx - 3
    ma60_prev = _safe_float(df.iloc[j].get("ma60", ma60), ma60) if j >= 0 else ma60
    ma60_slope = (ma60 - ma60_prev) / max(close, 1e-9)

    spread_sum = abs(ma20 - ma60) + abs(ma60 - ma120) + abs(ma20 - ma120)
    spread_norm = spread_sum / (3.0 * max(close, 1e-9))
    converge = 1.0 / (1.0 + spread_norm * 55.0)

    win = min(bar_idx + 1, 80)
    sl = max(0, bar_idx - win + 1)
    recent_high = float(df.iloc[sl : bar_idx + 1]["high"].max())
    near_high = 1.0 if close >= recent_high * 0.994 else 0.0

    scores = np.zeros(3, dtype=np.float64)

    in_pos = portfolio.unit > 1e-12
    bp = portfolio.buy_price if in_pos else 0.0
    unreal = (close - bp) / max(bp, 1e-9) if in_pos and bp > 0 else 0.0

    # 캔들 우선 규칙 — 장대 양봉/음봉 (policy.md)
    # 단순 근사: 바디가 종가 대비 일정 비율 이상이면 "장대"로 보고 강한 편향을 부여
    body = close - open_
    body_abs_norm = abs(body) / max(close, 1e-9)
    # 고가/저가가 있는 경우, 윗/아랫꼬리 포함 변동도 참고
    range_norm = (high - low) / max(close, 1e-9) if high >= low else 0.0
    big_candle = (body_abs_norm >= 0.012) or (range_norm >= 0.02)
    if big_candle:
        if body > 0:  # 장대 양봉
            scores[ACTION_BUY] += 1.1
            scores[ACTION_HOLD] -= 0.2
            if in_pos:
                scores[ACTION_SELL] -= 0.25
        elif body < 0:  # 장대 음봉
            scores[ACTION_SELL] += 1.1
            scores[ACTION_HOLD] -= 0.2
            if not in_pos:
                scores[ACTION_BUY] -= 0.35

    # Rule 1 — 왕복 수수료 대비 너무 얕은 익절 구간
    if in_pos and 0.0 < unreal < 0.01:
        scores[ACTION_HOLD] += 1.6
        scores[ACTION_SELL] -= 0.9
    elif in_pos and unreal >= 0.01:
        scores[ACTION_SELL] += 0.55
    elif in_pos and unreal < 0.0:
        scores[ACTION_SELL] += 0.45
        scores[ACTION_HOLD] -= 0.15

    # MA60 방향 (rules 4–7)
    if ma60_slope < -5e-6:
        scores[ACTION_BUY] -= 1.25
        scores[ACTION_SELL] += 0.55
        scores[ACTION_HOLD] += 0.35
    elif ma60_slope > 5e-6:
        scores[ACTION_BUY] += 0.65
        scores[ACTION_HOLD] += 0.25
        if in_pos:
            scores[ACTION_SELL] -= 0.15

    # 고점 수렴 — 매수 주의 (rule 3)
    if converge > 0.62 and near_high > 0.5:
        scores[ACTION_BUY] -= 1.15
        scores[ACTION_HOLD] += 0.45
        scores[ACTION_SELL] += 0.15

    # 에너지 부족(수렴) — 작은 변동 (rules 2, 10)
    if converge > 0.58:
        scores[ACTION_HOLD] += 0.5
        scores[ACTION_BUY] -= 0.35

    # 에너지 큼(발산) — 변동성·캔들 판단 강조: 과매수 과열 시 익절·손절 민감 (rules 10–11)
    if converge < 0.35:
        scores[ACTION_HOLD] -= 0.1
        scores[ACTION_SELL] += 0.25
        if in_pos and unreal > 0.005:
            scores[ACTION_SELL] += 0.2

    # 무포지션: 추세+에너지에 맞춰 진입만 허용
    if not in_pos:
        scores[ACTION_HOLD] += 0.35
        if ma60_slope > 5e-6 and not (converge > 0.6 and near_high > 0.5):
            scores[ACTION_BUY] += 0.75
        if ma60_slope < -5e-6:
            scores[ACTION_BUY] -= 0.5

    # 극단 라벨
    if "ext_crash_signal" in row.index and _bool01(row.get("ext_crash_signal")):
        scores[ACTION_SELL] += 0.85
        scores[ACTION_BUY] -= 0.5
    if "ext_surge_signal" in row.index and _bool01(row.get("ext_surge_signal")):
        scores[ACTION_BUY] += 0.35
        if in_pos and unreal > 0.02:
            scores[ACTION_SELL] += 0.25

    return scores.astype(np.float32)


def kl_student_from_teacher_logits(
    policy_logits_1d: torch.Tensor,
    teacher_logits_np: np.ndarray,
    *,
    temperature: float,
) -> torch.Tensor:
    """KL( teacher || student ) 한 스텝."""
    t = torch.as_tensor(teacher_logits_np, dtype=policy_logits_1d.dtype, device=policy_logits_1d.device)
    log_q = F.log_softmax(policy_logits_1d, dim=-1)
    temp = max(float(temperature), 1e-6)
    p = F.softmax(t / temp, dim=-1)
    return (p * (torch.log(p + 1e-8) - log_q)).sum()