예제 소스
차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 59개
train_png_policy_from_action_csv.py
"""
``ACTION_*.CSV`` (열 ``action``: 0=매도, 1=매수, 2=유지) 를 teacher 로 하여
차트 PNG 입력 ``PngToStatusPolicy`` 를 **교차엔트로피 모방학습** 한다.
뷰어 파이프라인과 동일:
- ``Market.get_data()`` + ``RL_MAX_BARS`` + ``merge_extreme_labels_from_csv``
- ``ChartImageGenerator.render_and_slice_by_x`` (``main_llm_db_test_ollama`` 상수)
- 봉 ``bar`` 직전까지 CSV 액션으로 ``PaperLlmPortfolio`` 를 재생하여 사이드 특징 일치
주의:
- 실제 VLM(LLM) 가중치를 고치는 것이 아니라, **PNG→액션 CNN 정책**만 학습한다.
- Kaleido 렌더가 느리면 ``--stride`` / ``--max-samples`` 로 줄일 것.
- ``len(ACTION CSV) != len(df)`` 이면 긴 쪽을 **tail** 기준으로 잘라 맞춘다(``RL_MAX_BARS`` tail 과 동일 가정).
실행 예:
python train_png_policy_from_action_csv.py \\
--actions cache/ACTION_repaired.CSV \\
--out cache/png_status_policy_action_repaired.pt
설정 기본값은 ``config/llm_rl_train.py`` 와 동일하게 ``import`` 한다.
"""
from __future__ import annotations
import argparse
import logging
import random
import sys
from pathlib import Path
import numpy as np
import pandas as pd
import torch
import torch.nn.functional as F
import torch.optim as optim
import main_llm_db_test_ollama as ollama_bt
from ChartImageGenerator import ChartImageConfig, ChartImageGenerator
from Market import Market
from config import llm_rl_train as tr
from lib.rl_data_labels import merge_extreme_labels_from_csv
from lib.rl_policy_teacher import write_rl_training_meta
from PngToStatus import (
DEFAULT_INPUT_H,
DEFAULT_INPUT_W,
load_png_status_policy,
save_png_status_policy,
)
logger = logging.getLogger(__name__)
def _load_df_like_viewer():
market = Market()
df = market.get_data()
if df is None or len(df) < tr.MIN_BARS + tr.BAR_STEP:
raise SystemExit("데이터가 부족합니다.")
n_full = len(df)
bar_offset = 0
cap = getattr(tr, "RL_MAX_BARS", None)
if cap is not None and n_full > int(cap):
bar_offset = n_full - int(cap)
df = df.iloc[-int(cap) :].reset_index(drop=True)
logger.info(
"RL_MAX_BARS=%s: tail %s봉 (전체 %s봉, bar_offset=%s)",
cap,
len(df),
n_full,
bar_offset,
)
df = merge_extreme_labels_from_csv(df, tr.LABELS_CSV_PATH, source_bar_offset=bar_offset)
df = ollama_bt._ensure_pb_mfi10(df)
return df
def _load_actions(path: Path, n: int) -> np.ndarray:
s = pd.read_csv(path, usecols=["action"]).squeeze("columns")
arr = s.astype(np.float64).round().astype(np.int64).to_numpy()
if len(arr) < n:
raise SystemExit(
f"액션 CSV 행({len(arr)})이 df 행({n})보다 짧습니다. "
"동일 구간으로 ACTION 을 만든 뒤 다시 실행하세요."
)
if len(arr) > n:
logger.warning("액션 CSV가 df보다 김 → tail %s행만 사용", n)
arr = arr[-n:]
return arr
def _side_from_portfolio(port: ollama_bt.PaperLlmPortfolio, price: float) -> tuple[bool, float, float]:
in_pos = port.unit > 1e-12
bp = float(getattr(port, "buy_price", 0.0))
unreal = (price - bp) / max(bp, 1e-9) if in_pos and bp > 0 else 0.0
pv = port.balance + port.unit * float(price)
cash_ratio = port.balance / max(pv, 1e-9)
return in_pos, float(unreal), float(cash_ratio)
def _render_png(chart_gen: ChartImageGenerator, df, bar: int) -> bytes:
df_hist = df.iloc[: bar + 1].copy()
chart_idx = len(df_hist)
base_png, _parts = chart_gen.render_and_slice_by_x(
df_hist,
idx=chart_idx,
max_data=min(ollama_bt.CHART_MAX_DATA, len(df_hist)),
ticker=ollama_bt.TICKER,
n_slices=ollama_bt.CHART_N_SLICES,
overlap_px=ollama_bt.CHART_OVERLAP_PX,
scale=ollama_bt.CHART_SCALE,
)
return base_png
def _collect_samples(
df,
chart_gen: ChartImageGenerator,
actions: np.ndarray,
*,
min_bar: int,
stride: int,
max_samples: int | None,
) -> list[tuple[bytes, np.ndarray, int]]:
"""각 원소: (png_bytes, side_vec float32 shape (3,), target_action int)."""
n = len(df)
port = ollama_bt.PaperLlmPortfolio(float(tr.INITIAL_KRW))
port.min_trade_bars = max(0, int(tr.MIN_TRADE_BARS))
out: list[tuple[bytes, np.ndarray, int]] = []
mb = max(50, int(min_bar))
st = max(1, int(stride))
for bar in range(n):
price = float(df.iloc[bar]["close"])
y = int(actions[bar])
if y not in (0, 1, 2):
raise SystemExit(f"bar={bar} 잘못된 액션: {y}")
if bar >= mb and (bar - mb) % st == 0:
if max_samples is None or len(out) < int(max_samples):
png = _render_png(chart_gen, df, bar)
in_pos, unreal, cr = _side_from_portfolio(port, price)
side = np.array(
[
1.0 if in_pos else 0.0,
float(np.clip(unreal, -1.0, 1.0)),
float(np.clip(cr, 0.0, 1.0)),
],
dtype=np.float32,
)
out.append((png, side, y))
sig, probs = ollama_bt._opinion_to_signal(
"매도" if y == 0 else "매수" if y == 1 else "유지"
)
port.execute_llm_trade(sig, price, current_bar=bar, probabilities=probs)
return out
def _train_loop(
model,
samples: list[tuple[bytes, np.ndarray, int]],
*,
epochs: int,
batch_size: int,
lr: float,
weight_decay: float,
device: torch.device,
) -> None:
model.train()
opt = optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)
idxs = list(range(len(samples)))
for ep in range(int(epochs)):
random.shuffle(idxs)
total_loss = 0.0
n_batches = 0
for start in range(0, len(idxs), max(1, int(batch_size))):
chunk = idxs[start : start + int(batch_size)]
xs: list[torch.Tensor] = []
sides: list[torch.Tensor] = []
ys: list[int] = []
for j in chunk:
png, side, y = samples[j]
xs.append(model.png_bytes_to_tensor(png).squeeze(0))
sides.append(torch.tensor(side, dtype=torch.float32, device=device).unsqueeze(0))
ys.append(y)
xb = torch.stack(xs, dim=0)
sb = torch.cat(sides, dim=0)
logits = model(xb, sb)
tgt = torch.tensor(ys, dtype=torch.long, device=device)
loss = F.cross_entropy(logits, tgt)
opt.zero_grad()
loss.backward()
opt.step()
total_loss += float(loss.item())
n_batches += 1
avg = total_loss / max(n_batches, 1)
logger.info("epoch %s/%s CE_loss=%.5f n_samples=%s", ep + 1, epochs, avg, len(samples))
def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
p = argparse.ArgumentParser(description="ACTION CSV → PngToStatusPolicy 모방학습")
p.add_argument("--actions", default="cache/ACTION_repaired.CSV", type=Path)
p.add_argument("--out", default="cache/png_status_policy_action_repaired.pt", type=Path)
p.add_argument("--epochs", type=int, default=8)
p.add_argument("--batch-size", type=int, default=16)
p.add_argument("--stride", type=int, default=3, help="샘플링 간격 (뷰어 BAR_STEP 과 맞추려면 3)")
p.add_argument(
"--max-samples",
type=int,
default=12_000,
help="수집 상한 (0이면 stride 만 적용해 전 구간, 메모리 주의)",
)
p.add_argument("--seed", type=int, default=42)
p.add_argument("--no-meta", action="store_true", help="체크포인트 옆 train_meta.json 생략")
args = p.parse_args()
random.seed(args.seed)
np.random.seed(args.seed)
torch.manual_seed(args.seed)
if not args.actions.is_file():
raise SystemExit(f"액션 파일 없음: {args.actions}")
logger.info("데이터 로드 …")
df = _load_df_like_viewer()
actions = _load_actions(args.actions, len(df))
chart_gen = ChartImageGenerator(
ticker=ollama_bt.TICKER,
config=ChartImageConfig(show=False),
)
min_bar = max(50, int(tr.MIN_BARS))
max_samp = None if int(args.max_samples) <= 0 else int(args.max_samples)
logger.info(
"샘플 수집 (min_bar=%s stride=%s max_samples=%s) … kaleido 느릴 수 있음",
min_bar,
args.stride,
max_samp,
)
samples = _collect_samples(
df,
chart_gen,
actions,
min_bar=min_bar,
stride=int(args.stride),
max_samples=max_samp,
)
if len(samples) < 32:
raise SystemExit(f"샘플이 너무 적습니다: {len(samples)} — stride 줄이거나 max-samples 늘리세요.")
pts_h = int(getattr(tr, "PNG_STATUS_INPUT_H", DEFAULT_INPUT_H))
pts_w = int(getattr(tr, "PNG_STATUS_INPUT_W", DEFAULT_INPUT_W))
pts_hidden = int(getattr(tr, "PNG_STATUS_HIDDEN", 256))
model = load_png_status_policy(
args.out,
h=pts_h,
w=pts_w,
hidden=pts_hidden,
)
lr = float(getattr(tr, "PNG_STATUS_LR", tr.LEARNING_RATE))
wd = float(tr.WEIGHT_DECAY)
logger.info("학습 시작 … device=%s samples=%s", model.device, len(samples))
_train_loop(
model,
samples,
epochs=int(args.epochs),
batch_size=int(args.batch_size),
lr=lr,
weight_decay=wd,
device=model.device,
)
save_png_status_policy(model, args.out)
logger.info("저장: %s", args.out.resolve())
if not args.no_meta:
write_rl_training_meta(
args.out,
policy_doc_path=getattr(tr, "POLICY_DOC_PATH", None),
extra={
"trainer": "********",
"action_csv": str(args.actions.resolve()),
"n_samples": len(samples),
"stride": int(args.stride),
"epochs": int(args.epochs),
"min_bar": min_bar,
},
)
print(f"[OK] 완료: {args.out.resolve()} samples={len(samples)}", flush=True)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("중단됨.", file=sys.stderr)
sys.exit(130)