예제 소스
차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 59개
build_action_from_labels.py
"""
``cache/market_extreme_labels.csv`` 를 읽어 ``ACTION_80.0_10.0.CSV`` 형식
(헤더 ``action``, 이후 행마다 ``0.0`` / ``1.0`` / ``2.0`` = 매도/매수/유지) CSV 를 만든다.
중요:
- 극단 라벨 CSV에는 "정답 매매 액션"이 없다. 이 스크립트는 **명시적인 규칙**으로
0/1/2 를 채울 뿐이며, 기존 ``ACTION_*.CSV`` 와 바이트 단위로 같다고 보장할 수 없다.
(파일명의 80.0_10.0 등이 무엇을 의미하는지 저장소에 정의가 없음.)
- 행 수는 **입력 라벨 CSV와 동일**하다. DB/빌드 시점이 다르면 ``ACTION_*.CSV`` 행 수와 다를 수 있다.
모드:
- ``regime``: ``ext_regime_id`` / 신호 플래그만으로 단순 매핑 (빠름).
- ``teacher``: ``lib.rl_policy_teacher.teacher_logits_from_chart`` 로짓의 argmax.
라벨의 ``ext_*`` 가 로짓에 반영된다. ``--walk`` 시 모의 포트폴리오를 봉마다 갱신.
실행 예:
python build_action_from_labels.py -o cache/ACTION_from_labels_teacher.CSV --mode teacher --walk
python build_action_from_labels.py -o cache/ACTION_from_labels_regime.CSV --mode regime
손실 매도 보정 (모의 체결 ``last_sell_profit_rate`` < 0 인 회차):
``--repair-pnl skip-buy``: 해당 포지션의 **매수 봉** 액션을 유지로 바꿔 진입 자체를 취소 (반복).
``--repair-pnl defer-sell``: 손실 매도 봉은 유지로 바꾸고, 수수료 감안 **목표가 이상**인
뒤쪽 봉 중 원래 매수가 아닌 첫 봉에 매도를 두어 익절을 노림. 목표 봉이 없으면 skip-buy 와 동일.
과거 데이터에 대한 사후 조정이라 **미래 가격을 본 탐색**이 포함되며, 실전과 다를 수 있음.
"""
from __future__ import annotations
import argparse
import logging
from pathlib import Path
import numpy as np
import pandas as pd
import main_llm_db_test_ollama as ollama_bt
from lib.rl_policy_teacher import (
ACTION_BUY,
ACTION_HOLD,
ACTION_SELL,
teacher_logits_from_chart,
)
def _action_from_regime_row(row: pd.Series) -> int:
"""극단 레짐·플래그 기반 휴리스틱 (단순)."""
crash = bool(row.get("ext_crash_signal"))
surge = bool(row.get("ext_surge_signal"))
if crash and surge:
return ACTION_HOLD
if crash:
return ACTION_SELL
if surge:
return ACTION_BUY
rid = int(row.get("ext_regime_id", 0) or 0)
if rid == 1:
return ACTION_SELL
if rid == 2:
return ACTION_BUY
if rid == 3:
return ACTION_HOLD
return ACTION_HOLD
def _actions_regime(df: pd.DataFrame) -> list[int]:
return [_action_from_regime_row(df.iloc[i]) for i in range(len(df))]
def _actions_teacher(
df: pd.DataFrame,
*,
walk: bool,
initial_krw: float,
min_trade_bars: int,
) -> list[int]:
out: list[int] = []
n = len(df)
port = ollama_bt.PaperLlmPortfolio(float(initial_krw))
port.min_trade_bars = max(0, int(min_trade_bars))
for bar in range(n):
if not walk:
port = ollama_bt.PaperLlmPortfolio(float(initial_krw))
port.min_trade_bars = max(0, int(min_trade_bars))
price = float(df.iloc[bar]["close"])
logits = teacher_logits_from_chart(df, bar, port, price)
act = int(np.argmax(logits))
out.append(act)
if walk:
sig, probs = ollama_bt._opinion_to_signal(
"매도" if act == ACTION_SELL else "매수" if act == ACTION_BUY else "유지"
)
port.execute_llm_trade(sig, price, current_bar=bar, probabilities=probs)
return out
def write_action_csv(path: Path, actions: list[int]) -> None:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8", newline="\n") as f:
f.write("action\n")
for a in actions:
f.write(f"{float(int(a))}\n")
def _simulate_closed_trades(
df: pd.DataFrame,
actions: list[int],
*,
initial_krw: float,
min_trade_bars: int,
) -> list[dict]:
"""모의 체결 기준 완료된 매수→매도 회차. pnl_pct 는 ``last_sell_profit_rate`` (%)."""
port = ollama_bt.PaperLlmPortfolio(float(initial_krw))
port.min_trade_bars = max(0, int(min_trade_bars))
trades: list[dict] = []
entry_bar: int | None = None
buy_price = 0.0
n = len(df)
for bar in range(n):
price = float(df.iloc[bar]["close"])
act = int(actions[bar])
sig, probs = ollama_bt._opinion_to_signal(
"매도" if act == ACTION_SELL else "매수" if act == ACTION_BUY else "유지"
)
executed = port.execute_llm_trade(sig, price, current_bar=bar, probabilities=probs)
if executed and sig == 1:
entry_bar = bar
buy_price = float(port.buy_price)
elif executed and sig == 0:
if entry_bar is not None:
trades.append(
{
"entry_bar": entry_bar,
"exit_bar": bar,
"pnl_pct": float(port.last_sell_profit_rate),
"buy_price": buy_price,
}
)
entry_bar = None
return trades
def _repair_pnl_skip_buy(
actions: list[int],
df: pd.DataFrame,
*,
initial_krw: float,
min_trade_bars: int,
max_passes: int,
) -> tuple[list[int], int]:
out = list(actions)
n_fix = 0
for _ in range(max_passes):
trades = _simulate_closed_trades(
df, out, initial_krw=initial_krw, min_trade_bars=min_trade_bars
)
neg = [t for t in trades if t["pnl_pct"] < 0]
if not neg:
break
worst = min(neg, key=lambda t: t["pnl_pct"])
eb = int(worst["entry_bar"])
if int(out[eb]) != ACTION_BUY:
break
out[eb] = ACTION_HOLD
n_fix += 1
return out, n_fix
def _repair_pnl_defer_sell(
actions: list[int],
df: pd.DataFrame,
*,
initial_krw: float,
min_trade_bars: int,
max_passes: int,
) -> tuple[list[int], int]:
fee = float(ollama_bt.PaperLlmPortfolio.BASIC_FEES)
target_mul = 1.0 + 2.0 * fee + 0.002
out = list(actions)
n_fix = 0
for _ in range(max_passes):
trades = _simulate_closed_trades(
df, out, initial_krw=initial_krw, min_trade_bars=min_trade_bars
)
neg = [t for t in trades if t["pnl_pct"] < 0]
if not neg:
break
worst = min(neg, key=lambda t: t["pnl_pct"])
B = int(worst["entry_bar"])
S = int(worst["exit_bar"])
buy_px = float(worst["buy_price"])
target = buy_px * target_mul
if int(out[S]) != ACTION_SELL:
out[B] = ACTION_HOLD
n_fix += 1
continue
T: int | None = None
for bar in range(S + 1, len(df)):
if float(df.iloc[bar]["close"]) < target:
continue
if int(out[bar]) == ACTION_BUY:
continue
T = bar
break
if T is None:
out[B] = ACTION_HOLD
n_fix += 1
continue
out[S] = ACTION_HOLD
for bar in range(S + 1, T):
out[bar] = ACTION_HOLD
out[T] = ACTION_SELL
n_fix += 1
return out, n_fix
def main() -> None:
parser = argparse.ArgumentParser(description="극단 라벨 CSV → ACTION 형식 CSV")
parser.add_argument(
"-i",
"--input",
default="cache/market_extreme_labels.csv",
help="입력 라벨 CSV (build_market_extreme_labels.py 출력)",
)
parser.add_argument(
"-o",
"--output",
required=True,
help="출력 ACTION CSV 경로",
)
parser.add_argument(
"--mode",
choices=("regime", "teacher"),
default="teacher",
help="regime: 레짐/플래그만, teacher: policy.md 근사 teacher 로짓 argmax",
)
parser.add_argument(
"--walk",
action="store_true",
help="teacher 모드에서만: 봉 순서대로 모의 포트를 유지하며 로짓 계산",
)
parser.add_argument("--initial-krw", type=float, default=5_000_000.0)
parser.add_argument("--min-trade-bars", type=int, default=5)
parser.add_argument(
"--repair-pnl",
choices=("none", "skip-buy", "defer-sell"),
default="none",
help="손실 매도(pnl%%<0) 회차를 줄이도록 액션 후처리 (사후·미래가 참조 가능)",
)
parser.add_argument(
"--repair-max-passes",
type=int,
default=50_000,
help="repair 루프 최대 회전 (손실 거래 수 상한에 가깝게)",
)
parser.add_argument("-q", "--quiet", action="store_true")
args = parser.parse_args()
logging.basicConfig(
level=logging.WARNING if args.quiet else logging.INFO,
format="%(levelname)s %(message)s",
)
inp = Path(args.input)
if not inp.is_file():
raise SystemExit(f"입력 파일 없음: {inp}")
df = pd.read_csv(inp)
if len(df) == 0:
raise SystemExit("입력 CSV 가 비었습니다.")
if args.mode == "regime":
actions = _actions_regime(df)
else:
actions = _actions_teacher(
df,
walk=bool(args.walk),
initial_krw=float(args.initial_krw),
min_trade_bars=int(args.min_trade_bars),
)
if args.repair_pnl != "none":
if args.repair_pnl == "skip-buy":
actions, n_fix = _repair_pnl_skip_buy(
actions,
df,
initial_krw=float(args.initial_krw),
min_trade_bars=int(args.min_trade_bars),
max_passes=int(args.repair_max_passes),
)
else:
actions, n_fix = _repair_pnl_defer_sell(
actions,
df,
initial_krw=float(args.initial_krw),
min_trade_bars=int(args.min_trade_bars),
max_passes=int(args.repair_max_passes),
)
if not args.quiet:
print(f"[repair-pnl:{args.repair_pnl}] 조정 스텝 수 ≈ {n_fix}", flush=True)
out = Path(args.output)
write_action_csv(out, actions)
print(f"[OK] 저장: {out.resolve()} rows={len(actions)} mode={args.mode}", flush=True)
if __name__ == "__main__":
main()