예제 소스
차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 59개
train_qwen_chart_lora.py
"""
모인 차트 PNG 로 ``models/Qwen2.5-VL-3B-Instruct`` LoRA 를 학습한다.
정답은 ``Account.py`` 와 같다.
- 종목 KRW-ETH, 한 번에 전량 매수·전량 매도
- 수수료 편도 0.0005
- 앞쪽 가격이 수수료를 넘는 목표에 먼저 닿으면 그 봉은 매수, 목표 봉은 매도, 그 사이는 유지
- 손절 폭에 먼저 닿으면 그 진입은 만들지 않는다
업비트 주문 API 는 호출하지 않는다. 가격은 테스트 CSV 종가, 차트는 ``chart/{bar//10000:04d}/{bar:07d}.png``.
실행:
python train_qwen_chart_lora.py
"""
from __future__ import annotations
import argparse
import logging
import random
import time
from datetime import datetime
from pathlib import Path
import numpy as np
import pandas as pd
import torch
from lib.chart_png import chart_png_path, existing_chart_bars
logger = logging.getLogger(__name__)
FEE = 0.0005 # Account.BASIC_FEES
TICKER = "KRW-ETH"
MODEL_DIR = Path("models/Qwen2.5-VL-3B-Instruct")
CHART_DIR = Path("chart")
LABEL_CSV = Path("cache/market_extreme_labels.csv")
def _breakeven_mul() -> float:
return 1.0 / ((1.0 - FEE) ** 2)
def build_actions(
close: np.ndarray,
*,
horizon: int,
net_profit: float,
stop_drop: float,
) -> np.ndarray:
"""0=매도, 1=매수, 2=유지. 포지션은 한 번에 하나만."""
n = len(close)
actions = np.full(n, 2, dtype=np.int64)
target = _breakeven_mul() * (1.0 + float(net_profit))
stop_px = 1.0 - float(stop_drop)
t = 0
while t < n - 1:
entry = float(close[t])
if entry <= 0:
t += 1
continue
end = min(n, t + 1 + int(horizon))
sell_at = None
for s in range(t + 1, end):
px = float(close[s])
if px <= entry * stop_px:
break
if px >= entry * target:
sell_at = s
break
if sell_at is None:
t += 1
continue
actions[t] = 1
actions[sell_at] = 0
t = sell_at + 1
return actions
def _net_return(entry: float, price: float) -> float:
if entry <= 0:
return 0.0
return price / entry * ((1.0 - FEE) ** 2) - 1.0
def _prompt(in_pos: bool, entry: float, price: float) -> str:
if in_pos:
net = _net_return(entry, price) * 100.0
state = f"ETH 보유 중. 진입가 {entry:,.0f}원. 수수료 반영 수익률 {net:+.2f}%."
else:
state = "ETH 보유 없음."
return (
f"업비트 {TICKER} 차트다. 주문은 시장가이고 수수료는 편도 0.05%다. "
"보유가 없을 때만 잔고의 99.5%를 매수하고, 보유 중이면 전량만 매도한다. "
"수수료를 넘기지 못하는 작은 하락에서는 매도하지 말고 유지한다. "
f"{state} 지금 취할 행동을 매수, 매도, 유지 중 한 단어로만 답하라."
)
def _answer(action: int) -> str:
if action == 1:
return "매수"
if action == 0:
return "매도"
return "유지"
def load_samples(
*,
horizon: int,
net_profit: float,
stop_drop: float,
chart_dir: Path,
csv_path: Path,
) -> list[dict]:
df = pd.read_csv(csv_path, usecols=["bar_index", "close"])
df["bar_index"] = df["bar_index"].astype(np.int64)
df["close"] = pd.to_numeric(df["close"], errors="coerce")
df = df.dropna().sort_values("bar_index").reset_index(drop=True)
actions = build_actions(
df["close"].to_numpy(dtype=np.float64),
horizon=horizon,
net_profit=net_profit,
stop_drop=stop_drop,
)
have = existing_chart_bars(chart_dir)
logger.info("차트 PNG %s장", len(have))
samples: list[dict] = []
in_pos = False
entry = 0.0
bars = df["bar_index"].to_numpy()
closes = df["close"].to_numpy(dtype=np.float64)
for i in range(len(df)):
bar = int(bars[i])
price = float(closes[i])
path = chart_png_path(chart_dir, bar)
action = int(actions[i])
if bar in have:
samples.append(
{
"path": str(path),
"prompt": _prompt(in_pos, entry, price),
"answer": _answer(action),
"action": action,
}
)
if action == 1:
in_pos = True
entry = price
elif action == 0:
in_pos = False
entry = 0.0
return samples
def _pick_device() -> torch.device:
if torch.cuda.is_available():
return torch.device("cuda")
if torch.backends.mps.is_available():
return torch.device("mps")
return torch.device("cpu")
def _draw(samples: list[dict], rng: random.Random) -> dict:
by = {0: [], 1: [], 2: []}
for s in samples:
by[int(s["action"])].append(s)
# 매수·매도를 유지보다 자주 보여 준다.
weights = []
pools = []
for k, w in ((1, 0.4), (0, 0.4), (2, 0.2)):
if by[k]:
pools.append(by[k])
weights.append(w)
pool = rng.choices(pools, weights=weights, k=1)[0]
return rng.choice(pool)
def train(args: argparse.Namespace) -> None:
from peft import LoraConfig, PeftModel, get_peft_model
from qwen_vl_utils import process_vision_info
from transformers import AutoProcessor, Qwen2_5_VLForConditionalGeneration
samples = load_samples(
horizon=args.horizon,
net_profit=args.net_profit,
stop_drop=args.stop_drop,
chart_dir=args.chart_dir,
csv_path=args.csv,
)
if len(samples) < 8:
raise SystemExit(f"학습 가능한 차트가 너무 적습니다: {len(samples)}")
counts = {name: sum(1 for s in samples if s["answer"] == name) for name in ("매수", "매도", "유지")}
logger.info("샘플 %s · %s", len(samples), counts)
device = _pick_device()
dtype = torch.bfloat16 if device.type in ("cuda", "mps") else torch.float32
logger.info("모델 로드: %s · %s · %s", args.model, device, dtype)
model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
str(args.model),
torch_dtype=dtype,
trust_remote_code=True,
)
processor = AutoProcessor.from_pretrained(str(args.model), trust_remote_code=True)
if hasattr(processor, "image_processor"):
processor.image_processor.max_pixels = int(args.max_pixels)
processor.image_processor.min_pixels = 256 * 28 * 28
model = model.to(device)
model.enable_input_require_grads()
model.gradient_checkpointing_enable()
model.config.use_cache = False
adapter = Path(args.out)
if (adapter / "adapter_config.json").is_file():
logger.info("기존 LoRA 이어서 학습: %s", adapter)
model = PeftModel.from_pretrained(model, str(adapter), is_trainable=True)
else:
lora = LoraConfig(
r=int(args.lora_r),
lora_alpha=int(args.lora_alpha),
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM",
target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
)
model = get_peft_model(model, lora)
model.print_trainable_parameters()
model.train()
optim = torch.optim.AdamW(
(p for p in model.parameters() if p.requires_grad),
lr=float(args.lr),
)
rng = random.Random(args.seed)
out = Path(args.out)
out.mkdir(parents=True, exist_ok=True)
until_ts = _parse_until(getattr(args, "until", None))
last_step = 0
for step in range(1, int(args.steps) + 1):
if until_ts is not None and time.time() >= until_ts:
logger.info("시각 제한 도달 · step=%s/%s", step - 1, args.steps)
break
last_step = step
item = _draw(samples, rng)
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": item["path"]},
{"type": "text", "text": item["prompt"]},
],
}
]
prompt_text = processor.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
full_messages = messages + [
{"role": "assistant", "content": [{"type": "text", "text": item["answer"]}]}
]
full_text = processor.apply_chat_template(
full_messages, tokenize=False, add_generation_prompt=False
)
image_inputs, video_inputs = process_vision_info(messages)
prompt_inputs = processor(
text=[prompt_text],
images=image_inputs,
videos=video_inputs,
padding=True,
return_tensors="pt",
)
full_inputs = processor(
text=[full_text],
images=image_inputs,
videos=video_inputs,
padding=True,
return_tensors="pt",
)
prompt_len = int(prompt_inputs["input_ids"].shape[1])
full_inputs = {k: v.to(device) for k, v in full_inputs.items()}
labels = full_inputs["input_ids"].clone()
if prompt_len >= labels.shape[1]:
logger.warning("프롬프트가 전체 토큰보다 길어 이 스텝을 건너뜁니다")
continue
labels[:, :prompt_len] = -100
labels[labels == processor.tokenizer.pad_token_id] = -100
outputs = model(**full_inputs, labels=labels)
loss = outputs.loss
optim.zero_grad(set_to_none=True)
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optim.step()
if step == 1 or step % 10 == 0 or step == int(args.steps):
logger.info("step %s/%s loss=%.4f answer=%s", step, args.steps, float(loss.item()), item["answer"])
if step % int(args.save_every) == 0 or step == int(args.steps):
model.save_pretrained(out)
processor.save_pretrained(out)
logger.info("LoRA 저장: %s", out.resolve())
if last_step > 0 and last_step % int(args.save_every) != 0 and last_step != int(args.steps):
model.save_pretrained(out)
processor.save_pretrained(out)
logger.info("LoRA 저장: %s", out.resolve())
print(f"[OK] LoRA 저장: {out.resolve()} samples={len(samples)} steps={last_step}", flush=True)
def _parse_until(value: str | None) -> float | None:
if not value:
return None
text = str(value).strip()
try:
return datetime.fromisoformat(text).timestamp()
except ValueError as exc:
raise SystemExit(f"--until 시각을 해석할 수 없습니다: {value}") from exc
def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
p = argparse.ArgumentParser(description="Qwen2.5-VL 차트 LoRA")
p.add_argument("--model", type=Path, default=MODEL_DIR)
p.add_argument("--csv", type=Path, default=LABEL_CSV)
p.add_argument("--chart-dir", type=Path, default=CHART_DIR)
p.add_argument("--out", type=Path, default=Path("cache/qwen_chart_lora"))
p.add_argument("--steps", type=int, default=200)
p.add_argument("--save-every", type=int, default=50)
p.add_argument("--lr", type=float, default=1e-4)
p.add_argument("--lora-r", type=int, default=8)
p.add_argument("--lora-alpha", type=int, default=16)
p.add_argument("--horizon", type=int, default=180)
p.add_argument("--net-profit", type=float, default=0.003, help="수수료를 넘는 추가 수익률")
p.add_argument("--stop-drop", type=float, default=0.015, help="이 하락이 목표보다 먼저면 매수하지 않음")
p.add_argument("--max-pixels", type=int, default=512 * 512)
p.add_argument("--seed", type=int, default=7)
p.add_argument(
"--until",
type=str,
default=None,
help="이 로컬 시각(ISO)까지 학습. 예: 2026-10-21T20:00:00",
)
args = p.parse_args()
if not args.model.is_dir():
raise SystemExit(f"모델 경로 없음: {args.model}")
train(args)
if __name__ == "__main__":
main()