c3334e4f77
deepcoin 패키지를 bithumb으로 rename하고, 3단계 live 운영·사이징 튜닝·텔레그램 알림을 통합한다. Co-authored-by: Cursor <cursoragent@cursor.com>
98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
"""연속 매수·매도 클러스터 상태별 매수·매도 비율 규칙."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
CLUSTER_SIZE_KEYS: tuple[str, ...] = ("1", "2", "3+")
|
|
|
|
|
|
def cluster_size_key(cluster_size: int) -> str:
|
|
"""클러스터 크기를 규칙 lookup 키로 변환한다."""
|
|
size = max(int(cluster_size), 1)
|
|
if size >= 3:
|
|
return "3+"
|
|
return str(size)
|
|
|
|
|
|
def resolve_buy_cash_pct(
|
|
rules: dict[str, Any] | None,
|
|
cluster_size: int,
|
|
default: float | None,
|
|
) -> float | None:
|
|
"""클러스터 상태에 맞는 1회 매수 현금 비율을 반환한다."""
|
|
if not rules:
|
|
return default
|
|
by_side = (rules.get("by_cluster") or {}).get("buy") or {}
|
|
key = cluster_size_key(cluster_size)
|
|
if key in by_side:
|
|
return float(by_side[key])
|
|
fallback = rules.get("default_buy_cash_pct")
|
|
if fallback is not None:
|
|
return float(fallback)
|
|
return default
|
|
|
|
|
|
def resolve_sell_coin_pct(
|
|
rules: dict[str, Any] | None,
|
|
cluster_size: int,
|
|
default: float | None,
|
|
) -> float | None:
|
|
"""클러스터 상태에 맞는 1회 매도 코인 비율을 반환한다."""
|
|
if not rules:
|
|
return default
|
|
by_side = (rules.get("by_cluster") or {}).get("sell") or {}
|
|
key = cluster_size_key(cluster_size)
|
|
if key in by_side:
|
|
return float(by_side[key])
|
|
fallback = rules.get("default_sell_coin_pct")
|
|
if fallback is not None:
|
|
return float(fallback)
|
|
return default
|
|
|
|
|
|
def load_sizing_rules(path: Path) -> dict[str, Any] | None:
|
|
"""JSON 사이징 규칙을 로드한다. 파일이 없으면 None."""
|
|
if not path.exists():
|
|
return None
|
|
with path.open(encoding="utf-8") as fp:
|
|
data = json.load(fp)
|
|
return data if isinstance(data, dict) else None
|
|
|
|
|
|
def save_sizing_rules(rules: dict[str, Any], path: Path) -> Path:
|
|
"""사이징 규칙 JSON 저장."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8") as fp:
|
|
json.dump(rules, fp, ensure_ascii=False, indent=2)
|
|
return path
|
|
|
|
|
|
def empty_by_cluster() -> dict[str, dict[str, float]]:
|
|
"""빈 by_cluster 구조."""
|
|
return {"buy": {}, "sell": {}}
|
|
|
|
|
|
def merge_rules(
|
|
*,
|
|
default_buy: float,
|
|
default_sell: float,
|
|
by_cluster: dict[str, dict[str, float]] | None = None,
|
|
technique_id: str = "",
|
|
symbol: str = "",
|
|
tuning: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""튜닝 결과 dict를 표준 규칙 형식으로 만든다."""
|
|
return {
|
|
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"technique_id": technique_id,
|
|
"symbol": symbol,
|
|
"default_buy_cash_pct": round(float(default_buy), 4),
|
|
"default_sell_coin_pct": round(float(default_sell), 4),
|
|
"by_cluster": by_cluster or empty_by_cluster(),
|
|
"tuning": tuning or {},
|
|
}
|