예제 소스
차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 59개
evaluate_qwen_chart_lora.py
"""
Qwen 차트 LoRA를 PNG 300번부터 마지막 이미지까지 순차 평가한다.
실제 ``Account.py``의 시장가 운용 규칙을 네트워크 호출 없이 재현한다.
* 초기 원화 잔고의 99.5%를 매수하고 매수 수수료 0.05%를 반영
* 보유 중인 코인은 전량 매도하고 매도 수수료 0.05%를 반영
* 이미 보유 중인 매수와 미보유 상태의 매도는 무시
평가는 모델 어댑터 해시별 디렉터리에 체크포인트와 거래 내역을 저장한다.
같은 모델로 다시 실행하면 마지막 완료 bar 다음부터 이어서 진행한다.
"""
from __future__ import annotations
import argparse
import csv
import hashlib
import json
import logging
import math
import random
import re
import time
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
import pandas as pd
import torch
from lib.chart_png import iter_chart_pngs
logger = logging.getLogger(__name__)
FEE = 0.0005 # Account.BASIC_FEES
BUY_BALANCE_RATIO = 0.995 # Account.unit_buy
MIN_ORDER_KRW = 5_000
DEFAULT_INITIAL_KRW = 5_000_000.0
DEFAULT_START_BAR = 300
MODEL_DIR = Path("models/Qwen2.5-VL-3B-Instruct")
ADAPTER_DIR = Path("cache/qwen_chart_lora")
CHART_DIR = Path("chart")
MARKET_CSV = Path("cache/market_extreme_labels.csv")
EVAL_ROOT = Path("cache/qwen_chart_lora/eval")
ACTION_NAMES = {0: "매도", 1: "매수", 2: "유지"}
def _net_return(entry: float, price: float) -> float:
if entry <= 0:
return 0.0
return price / entry * ((1.0 - FEE) ** 2) - 1.0
def build_prompt(in_position: bool, entry: float, price: float) -> str:
"""학습에 사용한 프롬프트와 동일한 평가 프롬프트."""
if in_position:
net = _net_return(entry, price) * 100.0
state = f"ETH 보유 중. 진입가 {entry:,.0f}원. 수수료 반영 수익률 {net:+.2f}%."
else:
state = "ETH 보유 없음."
return (
"업비트 KRW-ETH 차트다. 주문은 시장가이고 수수료는 편도 0.05%다. "
"보유가 없을 때만 잔고의 99.5%를 매수하고, 보유 중이면 전량만 매도한다. "
"수수료를 넘기지 못하는 작은 하락에서는 매도하지 말고 유지한다. "
f"{state} 지금 취할 행동을 매수, 매도, 유지 중 한 단어로만 답하라."
)
def parse_action(text: str) -> tuple[int, bool]:
"""모델 응답을 action으로 변환. 해석 실패는 유지로 처리한다."""
cleaned = (text or "").strip()
if cleaned in ("매수", "매도", "유지"):
return {"매도": 0, "매수": 1, "유지": 2}[cleaned], True
matches = re.findall(r"매수|매도|유지", cleaned)
if matches:
return {"매도": 0, "매수": 1, "유지": 2}[matches[0]], True
return 2, False
@dataclass
class PaperAccount:
"""``Account.py``의 매수·매도 규칙을 재현하는 오프라인 계좌."""
initial_krw: float = DEFAULT_INITIAL_KRW
balance: float = DEFAULT_INITIAL_KRW
unit: float = 0.0
buy_price: float = 0.0
buy_cost: float = 0.0
completed_trades: int = 0
profitable_trades: int = 0
buy_count: int = 0
sell_count: int = 0
realized_profit: float = 0.0
@classmethod
def create(cls, initial_krw: float) -> "PaperAccount":
amount = float(initial_krw)
return cls(initial_krw=amount, balance=amount)
def portfolio_value(self, price: float) -> float:
return self.balance + self.unit * float(price)
def liquidation_value(self, price: float) -> float:
return self.balance + self.unit * float(price) * (1.0 - FEE)
def execute(self, action: int, price: float) -> bool:
price = float(price)
if action == 1 and self.unit <= 0:
buy_amount = math.floor(self.balance * BUY_BALANCE_RATIO)
if buy_amount < MIN_ORDER_KRW:
return False
self.unit = (buy_amount / price) * (1.0 - FEE)
self.balance -= buy_amount
self.buy_price = price
self.buy_cost = float(buy_amount)
self.buy_count += 1
return True
if action == 0 and self.unit > 0:
proceeds = self.unit * price * (1.0 - FEE)
profit = proceeds - self.buy_cost
self.balance += proceeds
self.realized_profit += profit
self.completed_trades += 1
if profit > 0:
self.profitable_trades += 1
self.unit = 0.0
self.buy_price = 0.0
self.buy_cost = 0.0
self.sell_count += 1
return True
return False
def adapter_fingerprint(adapter_dir: Path) -> str:
path = adapter_dir / "adapter_model.safetensors"
if not path.is_file():
raise FileNotFoundError(f"LoRA 가중치 없음: {path}")
digest = hashlib.sha256()
with path.open("rb") as fh:
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _load_available_bars(
chart_dir: Path,
csv_path: Path,
*,
start_bar: int,
end_bar: int | None,
max_bars: int | None,
) -> list[tuple[int, float, Path]]:
prices = pd.read_csv(csv_path, usecols=["bar_index", "close"])
prices["bar_index"] = pd.to_numeric(prices["bar_index"], errors="coerce")
prices["close"] = pd.to_numeric(prices["close"], errors="coerce")
prices = prices.dropna().set_index("bar_index")["close"]
price_by_bar = {int(k): float(v) for k, v in prices.items()}
rows: list[tuple[int, float, Path]] = []
for path in iter_chart_pngs(chart_dir):
if not path.stem.isdigit() or path.stat().st_size <= 0:
continue
bar = int(path.stem)
if bar < int(start_bar) or (end_bar is not None and bar > int(end_bar)):
continue
price = price_by_bar.get(bar)
if price is not None and math.isfinite(price) and price > 0:
rows.append((bar, price, path))
rows.sort(key=lambda item: item[0])
if max_bars is not None and int(max_bars) > 0:
rows = rows[: int(max_bars)]
return rows
def _load_model(model_dir: Path, adapter_dir: Path, max_pixels: int):
from peft import PeftModel
from transformers import AutoProcessor, Qwen2_5_VLForConditionalGeneration
if torch.cuda.is_available():
device = torch.device("cuda")
elif torch.backends.mps.is_available():
device = torch.device("mps")
else:
device = torch.device("cpu")
dtype = torch.bfloat16 if device.type in ("cuda", "mps") else torch.float32
logger.info("기본 모델 로드: %s · device=%s · dtype=%s", model_dir, device, dtype)
base = Qwen2_5_VLForConditionalGeneration.from_pretrained(
str(model_dir),
torch_dtype=dtype,
trust_remote_code=True,
).to(device)
model = PeftModel.from_pretrained(base, str(adapter_dir), is_trainable=False)
model.eval()
model.config.use_cache = True
processor = AutoProcessor.from_pretrained(str(adapter_dir), trust_remote_code=True)
if hasattr(processor, "image_processor"):
processor.image_processor.max_pixels = int(max_pixels)
processor.image_processor.min_pixels = 256 * 28 * 28
return model, processor, device
def infer_action(
*,
model: Any,
processor: Any,
device: torch.device,
image_path: Path,
prompt: str,
max_new_tokens: int,
) -> tuple[int, str, bool]:
from qwen_vl_utils import process_vision_info
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": str(image_path)},
{"type": "text", "text": prompt},
],
}
]
text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
image_inputs, video_inputs = process_vision_info(messages)
inputs = processor(
text=[text],
images=image_inputs,
videos=video_inputs,
padding=True,
return_tensors="pt",
)
inputs = {k: v.to(device) for k, v in inputs.items()}
with torch.inference_mode():
generated = model.generate(
**inputs,
max_new_tokens=max(2, int(max_new_tokens)),
do_sample=False,
)
prompt_len = int(inputs["input_ids"].shape[1])
answer = processor.batch_decode(
generated[:, prompt_len:],
skip_special_tokens=True,
clean_up_tokenization_spaces=False,
)[0].strip()
action, valid = parse_action(answer)
return action, answer, valid
def _atomic_json(path: Path, payload: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
tmp.replace(path)
def _restore_account(data: dict[str, Any], initial_krw: float) -> PaperAccount:
fields = set(PaperAccount.__dataclass_fields__)
values = {k: v for k, v in data.items() if k in fields}
values.setdefault("initial_krw", float(initial_krw))
return PaperAccount(**values)
def _truncate_predictions(path: Path, last_bar: int) -> None:
"""state보다 앞선 행만 남겨 중단 직후 재개 시 CSV 중복을 막는다."""
if not path.is_file():
return
tmp = path.with_suffix(path.suffix + ".tmp")
with path.open("r", encoding="utf-8", newline="") as source:
reader = csv.DictReader(source)
if not reader.fieldnames:
path.unlink(missing_ok=True)
return
with tmp.open("w", encoding="utf-8", newline="") as target:
writer = csv.DictWriter(target, fieldnames=reader.fieldnames)
writer.writeheader()
for row in reader:
try:
bar = int(row["bar_index"])
except (KeyError, TypeError, ValueError):
continue
if bar <= last_bar:
writer.writerow(row)
tmp.replace(path)
def _duration_expired(started: float, max_seconds: float | None, now: float) -> bool:
"""설정한 평가 시간이 지난 경우에만 참을 반환한다."""
return max_seconds is not None and max_seconds > 0 and now - started >= max_seconds
def evaluate(args: argparse.Namespace) -> Path:
started = time.time()
fingerprint = adapter_fingerprint(args.adapter)
run_dir = args.output_root / fingerprint[:16]
state_path = run_dir / "state.json"
result_path = run_dir / "result.json"
trades_path = run_dir / "predictions.csv"
run_dir.mkdir(parents=True, exist_ok=True)
start_floor = int(args.start_bar)
rows = _load_available_bars(
args.chart_dir,
args.csv,
start_bar=start_floor,
end_bar=args.end_bar,
max_bars=None,
)
if not rows:
raise SystemExit("평가할 PNG와 CSV 가격 쌍이 없습니다.")
if args.random_start:
rng = random.Random(args.seed)
start_idx = rng.randrange(len(rows))
rows = rows[start_idx:]
args.start_bar = rows[0][0]
logger.info("랜덤 시작봉=%s (후보 %s개 중 index=%s)", args.start_bar, start_idx + len(rows), start_idx)
if args.max_bars is not None and int(args.max_bars) > 0:
rows = rows[: int(args.max_bars)]
eval_last_bar = rows[-1][0]
logger.info(
"평가 범위: start_bar=%s · last_bar=%s · available_images=%s · adapter=%s",
args.start_bar,
eval_last_bar,
len(rows),
fingerprint[:16],
)
account = PaperAccount.create(args.initial_krw)
action_counts = {"매도": 0, "매수": 0, "유지": 0}
executed_counts = {"매도": 0, "매수": 0}
invalid_outputs = 0
processed = 0
max_portfolio = float(args.initial_krw)
max_drawdown = 0.0
resume_after = -1
if state_path.is_file() and not args.force:
state = json.loads(state_path.read_text(encoding="utf-8"))
if state.get("adapter_sha256") != fingerprint:
raise RuntimeError("체크포인트의 LoRA 해시가 현재 어댑터와 다릅니다.")
if state.get("completed") and result_path.is_file():
logger.info("이미 완료된 동일 모델 평가: %s", result_path)
return result_path
account = _restore_account(state.get("account", {}), args.initial_krw)
action_counts.update(state.get("action_counts", {}))
executed_counts.update(state.get("executed_counts", {}))
invalid_outputs = int(state.get("invalid_outputs", 0))
processed = int(state.get("processed", 0))
max_portfolio = float(state.get("max_portfolio", args.initial_krw))
max_drawdown = float(state.get("max_drawdown_pct", 0.0))
resume_after = int(state.get("last_processed_bar", -1))
_truncate_predictions(trades_path, resume_after)
logger.info("평가 재개: bar>%s · processed=%s", resume_after, processed)
elif args.force:
state_path.unlink(missing_ok=True)
result_path.unlink(missing_ok=True)
trades_path.unlink(missing_ok=True)
pending = [row for row in rows if row[0] > resume_after]
if pending:
model, processor, device = _load_model(args.model, args.adapter, args.max_pixels)
else:
model = processor = device = None
write_header = not trades_path.is_file() or trades_path.stat().st_size == 0
out = trades_path.open("a", encoding="utf-8", newline="")
writer = csv.DictWriter(
out,
fieldnames=[
"bar_index",
"close",
"model_output",
"action",
"action_name",
"valid_output",
"executed",
"balance",
"unit",
"portfolio_value",
"return_pct",
],
)
if write_header:
writer.writeheader()
eval_t0 = time.perf_counter()
last_bar = resume_after
price_by_bar = {bar: price for bar, price, _ in rows}
last_price = price_by_bar.get(resume_after, rows[0][1])
time_limit_reached = False
try:
for local_index, (bar, price, image_path) in enumerate(pending, start=1):
if _duration_expired(eval_t0, args.max_seconds, time.perf_counter()):
time_limit_reached = True
logger.info(
"평가 시간 제한 도달 · %.1f초 · processed=%s · last_bar=%s",
time.perf_counter() - eval_t0,
processed,
last_bar,
)
break
prompt = build_prompt(account.unit > 0, account.buy_price, price)
action, answer, valid = infer_action(
model=model,
processor=processor,
device=device,
image_path=image_path,
prompt=prompt,
max_new_tokens=args.max_new_tokens,
)
action_name = ACTION_NAMES[action]
action_counts[action_name] += 1
if not valid:
invalid_outputs += 1
executed = account.execute(action, price)
if executed and action_name in executed_counts:
executed_counts[action_name] += 1
pv = account.portfolio_value(price)
max_portfolio = max(max_portfolio, pv)
if max_portfolio > 0:
drawdown = (max_portfolio - pv) / max_portfolio * 100.0
max_drawdown = max(max_drawdown, drawdown)
processed += 1
last_bar = bar
last_price = price
writer.writerow(
{
"bar_index": bar,
"close": price,
"model_output": answer,
"action": action,
"action_name": action_name,
"valid_output": int(valid),
"executed": int(executed),
"balance": account.balance,
"unit": account.unit,
"portfolio_value": pv,
"return_pct": (pv / account.initial_krw - 1.0) * 100.0,
}
)
if local_index == 1 or local_index % int(args.save_every) == 0:
out.flush()
elapsed = time.perf_counter() - eval_t0
rate = local_index / max(elapsed, 1e-9)
remaining = (len(pending) - local_index) / max(rate, 1e-9)
state = {
"version": 1,
"completed": False,
"adapter_sha256": fingerprint,
"model_dir": str(args.model.resolve()),
"adapter_dir": str(args.adapter.resolve()),
"csv_path": str(args.csv.resolve()),
"chart_dir": str(args.chart_dir.resolve()),
"start_bar": int(args.start_bar),
"evaluation_last_bar": eval_last_bar,
"available_images": len(rows),
"processed": processed,
"last_processed_bar": last_bar,
"action_counts": action_counts,
"executed_counts": executed_counts,
"invalid_outputs": invalid_outputs,
"account": asdict(account),
"max_portfolio": max_portfolio,
"max_drawdown_pct": max_drawdown,
"updated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
}
_atomic_json(state_path, state)
logger.info(
"평가 %s/%s · bar=%s · %.3f장/초 · ETA %.1f시간 · 수익률=%+.3f%%",
local_index,
len(pending),
bar,
rate,
remaining / 3600.0,
(pv / account.initial_krw - 1.0) * 100.0,
)
finally:
out.close()
final_pv = account.portfolio_value(last_price)
liquidation_pv = account.liquidation_value(last_price)
first_price = rows[0][1]
buy_hold_unit = (math.floor(args.initial_krw * BUY_BALANCE_RATIO) / first_price) * (1.0 - FEE)
buy_hold_cash = args.initial_krw - math.floor(args.initial_krw * BUY_BALANCE_RATIO)
buy_hold_liquidation = buy_hold_cash + buy_hold_unit * last_price * (1.0 - FEE)
evaluation_elapsed = time.perf_counter() - eval_t0
all_available_bars_processed = not pending or last_bar >= eval_last_bar
result = {
"version": 1,
"completed": all_available_bars_processed or time_limit_reached,
"stop_reason": "time_limit" if time_limit_reached else "all_available_bars",
"requested_max_seconds": args.max_seconds,
"evaluation_elapsed_seconds": evaluation_elapsed,
"adapter_sha256": fingerprint,
"model_dir": str(args.model.resolve()),
"adapter_dir": str(args.adapter.resolve()),
"csv_path": str(args.csv.resolve()),
"chart_dir": str(args.chart_dir.resolve()),
"start_bar": int(args.start_bar),
"first_evaluated_bar": rows[0][0],
"last_evaluated_bar": last_bar,
"evaluation_last_bar": eval_last_bar,
"available_images": len(rows),
"processed": processed,
"missing_images_in_range": max(0, eval_last_bar - int(args.start_bar) + 1 - len(rows)),
"initial_krw": account.initial_krw,
"final_close": last_price,
"final_portfolio_value": final_pv,
"final_liquidation_value": liquidation_pv,
"return_pct": (final_pv / account.initial_krw - 1.0) * 100.0,
"liquidation_return_pct": (liquidation_pv / account.initial_krw - 1.0) * 100.0,
"buy_hold_liquidation_return_pct": (buy_hold_liquidation / args.initial_krw - 1.0) * 100.0,
"max_drawdown_pct": max_drawdown,
"action_counts": action_counts,
"executed_counts": executed_counts,
"invalid_outputs": invalid_outputs,
"completed_trades": account.completed_trades,
"profitable_trades": account.profitable_trades,
"win_rate_pct": account.profitable_trades / max(account.completed_trades, 1) * 100.0,
"realized_profit": account.realized_profit,
"open_position": account.unit > 0,
"account": asdict(account),
"elapsed_seconds": time.time() - started,
"predictions_csv": str(trades_path.resolve()),
"completed_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
}
_atomic_json(result_path, result)
_atomic_json(state_path, result)
logger.info(
"평가 완료(%s) · bar=%s~%s · 청산 수익률=%+.3f%% · 거래=%s · 승률=%.2f%%",
result["stop_reason"],
rows[0][0],
last_bar,
result["liquidation_return_pct"],
account.completed_trades,
result["win_rate_pct"],
)
print(json.dumps(result, ensure_ascii=False, indent=2), flush=True)
return result_path
def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
parser = argparse.ArgumentParser(description="Qwen 차트 LoRA Account 규칙 백테스트")
parser.add_argument("--model", type=Path, default=MODEL_DIR)
parser.add_argument("--adapter", type=Path, default=ADAPTER_DIR)
parser.add_argument("--csv", type=Path, default=MARKET_CSV)
parser.add_argument("--chart-dir", type=Path, default=CHART_DIR)
parser.add_argument("--output-root", type=Path, default=EVAL_ROOT)
parser.add_argument("--start-bar", type=int, default=DEFAULT_START_BAR)
parser.add_argument(
"--random-start",
action="store_true",
help="--start-bar 이상에서 아무 PNG나 골라 그 봉부터 평가한다",
)
parser.add_argument("--seed", type=int, default=None, help="랜덤 시작봉 시드")
parser.add_argument("--end-bar", type=int, default=None)
parser.add_argument("--max-bars", type=int, default=None, help="스모크 테스트용 평가 수 제한")
parser.add_argument(
"--max-seconds",
type=float,
default=None,
help="모델 로드 후 평가 실행 시간 제한(초). 현재 추론을 마친 뒤 종료한다.",
)
parser.add_argument("--initial-krw", type=float, default=DEFAULT_INITIAL_KRW)
parser.add_argument("--max-pixels", type=int, default=512 * 512)
parser.add_argument("--max-new-tokens", type=int, default=4)
parser.add_argument("--save-every", type=int, default=25)
parser.add_argument("--force", action="store_true", help="동일 모델의 기존 평가를 삭제하고 재시작")
args = parser.parse_args()
for path, label in (
(args.model, "기본 모델"),
(args.adapter, "LoRA"),
(args.csv, "시장 CSV"),
(args.chart_dir, "차트 디렉터리"),
):
if not path.exists():
raise SystemExit(f"{label} 없음: {path}")
evaluate(args)
if __name__ == "__main__":
main()