joungmin/upbit-trader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add market regime filter and compound reinvestment

Browse Source 
- Add market_regime.py: BTC/ETH/SOL/XRP weighted 2h trend score
  Bull(≥+1.5%) / Neutral / Bear(<-1%) regime detection with 10min cache
- strategy.py: dynamic TREND/VOL thresholds based on current regime
  Bull: 3%/1.5x, Neutral: 5%/2.0x, Bear: 8%/3.5x
- price_collector.py: always include leader coins in price history
- trader.py: compound reinvestment (profit added to budget, floor at initial)
- notify.py: regime info in hourly report, P&L icons (✅/❌, 💚/🔴)
- main.py: hourly status at top-of-hour, filter positions held 1h+
- backtest.py: timestop/combo comparison modes

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
joungmin
2026-03-01 10:14:36 +09:00
parent 035b3e2f30
commit 83a229dd26
8 changed files with 423 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
backtest.py
View File

@@ -70,6 +70,8 @@ COMPARE_TOP50 = "--top50-cmp" in sys.argv # 종목수 + 거래대금
COMPARE_TREND = "--trend-cmp" in sys.argv # 추세 상한선 비교 모드
COMPARE_TICKER = "--ticker-cmp" in sys.argv # 종목 승률 필터 비교 모드
COMPARE_WF = "--walkforward-cmp" in sys.argv # walk-forward 필터 비교 모드
COMPARE_TIMESTOP = "--timestop-cmp" in sys.argv # 타임스탑 조건 비교 모드
COMPARE_COMBO = "--combo-cmp" in sys.argv # 추세+거래량 조합 비교 모드
if "--10m" in sys.argv:
DEFAULT_INTERVAL = "minute10"
@@ -574,6 +576,8 @@ def simulate(
volume_mult: float = VOLUME_MULT, # 거래대금 급증 기준 배수
wf_min_wr: float = 0.0, # walk-forward: 직전 N건 승률 임계값 (0=비활성)
wf_window: int = 5, # walk-forward: 승률 계산 윈도우 크기
time_stop_h: float = TIME_STOP_H, # 타임스탑 기준 시간 (0=비활성)
time_stop_gain: float = TIME_STOP_GAIN, # 타임스탑 최소 수익률
) -> list[dict]:
"""단일 종목 전략 시뮬레이션.
@@ -582,6 +586,7 @@ def simulate(
volume_mult : 진입 조건 — 전일 거래량 > N일 평균 × volume_mult
wf_min_wr : walk-forward 필터 — 직전 wf_window건 승률이 이 값 미만이면 진입 차단
윈도우가 채워지기 전(워밍업)에는 필터 미적용
time_stop_h : 타임스탑 기준 시간 (0 이하 = 비활성)
"""
_hard_stop = hard_stop if hard_stop is not None else stop_loss
trades: list[dict] = []
@@ -614,7 +619,7 @@ def simulate(
reason = "trailing_stop"
elif _hard_stop < 999 and drop_buy >= _hard_stop:
reason = "hard_stop"
elif elapsed >= TIME_STOP_H and pnl < TIME_STOP_GAIN:
elif time_stop_h > 0 and elapsed >= time_stop_h and pnl < time_stop_gain:
reason = "time_stop"
if reason:
@@ -851,6 +856,8 @@ def run_scenario(
daily_features: Optional[dict] = None, # 미리 계산된 daily_feat {ticker: df}
wf_min_wr: float = 0.0, # walk-forward 승률 임계값
wf_window: int = 5, # walk-forward 윈도우
time_stop_h: float = TIME_STOP_H, # 타임스탑 기준 시간 (0=비활성)
time_stop_gain: float = TIME_STOP_GAIN, # 타임스탑 최소 수익률
) -> list[dict]:
"""주어진 파라미터로 전체 종목 시뮬레이션 (데이터는 외부에서 주입).
@@ -868,6 +875,8 @@ def run_scenario(
"reentry": reentry_above_sell,
"wf_min_wr": round(wf_min_wr, 3),
"wf_window": wf_window,
"time_stop_h": time_stop_h,
"time_stop_gain": round(time_stop_gain, 4),
}
if use_db_cache:
@@ -894,6 +903,8 @@ def run_scenario(
volume_mult=volume_mult,
wf_min_wr=wf_min_wr,
wf_window=wf_window,
time_stop_h=time_stop_h,
time_stop_gain=time_stop_gain,
)
all_trades.extend(trades)
@@ -1531,8 +1542,142 @@ def main_ticker_filter_cmp(interval: str = DEFAULT_INTERVAL) -> None:
print(f" ✗ {row[0]:<12} 승률 {row[1]:.0f}%")
def main_timestop_cmp(interval: str = DEFAULT_INTERVAL) -> None:
"""타임스탑 조건 비교: 트레일링 1.5%+8h(현행) vs 10%+8h vs 10%+24h vs 10%+없음."""
cfg = INTERVAL_CONFIG[interval]
label = cfg["label"]
trend = cfg["trend"]
print(f"\n{'='*60}")
print(f" 타임스탑 조건 비교 | {MONTHS}개월 | 상위 {TOP_N}종목 | {label}")
print(f" (A) 트레일링 1.5% + 타임스탑 8h ← 현재 설정")
print(f" (B) 트레일링 10% + 타임스탑 8h")
print(f" (C) 트레일링 10% + 타임스탑 24h")
print(f" (D) 트레일링 10% + 타임스탑 없음")
print(f"{'='*60}")
print("\n▶ 종목 데이터 로드 중 (DB 캐시 우선)...")
tickers = get_top_tickers(TOP_N)
print(f" → {tickers}")
data = fetch_all_data(tickers, interval)
print(f" 사용 종목: {list(data.keys())}")
daily_features = {t: build_daily_features(f["daily"]) for t, f in data.items()}
common = dict(
reentry_above_sell=True, interval_cd=interval,
use_db_cache=True, daily_features=daily_features,
trend_min_gain=trend,
)
print("\n▶ A: 트레일링 1.5% + 타임스탑 8h (현재 설정) ...")
trades_a = run_scenario(data, stop_loss=0.015, time_stop_h=8, **common)
print(f" 완료 ({len(trades_a)}건)")
print("\n▶ B: 트레일링 10% + 타임스탑 8h ...")
trades_b = run_scenario(data, stop_loss=0.10, time_stop_h=8, **common)
print(f" 완료 ({len(trades_b)}건)")
print("\n▶ C: 트레일링 10% + 타임스탑 24h ...")
trades_c = run_scenario(data, stop_loss=0.10, time_stop_h=24, **common)
print(f" 완료 ({len(trades_c)}건)")
print("\n▶ D: 트레일링 10% + 타임스탑 없음 ...")
trades_d = run_scenario(data, stop_loss=0.10, time_stop_h=0, **common)
print(f" 완료 ({len(trades_d)}건)")
compare_report([
("1.5%+8h(현행)", trades_a),
("10%+8h", trades_b),
("10%+24h", trades_c),
("10%+타임없음", trades_d),
], title=f"트레일링 + 타임스탑 조합 비교 ({label})")
report(trades_a, f"A: 트레일링 1.5% + 타임스탑 8h | {label}")
report(trades_b, f"B: 트레일링 10% + 타임스탑 8h | {label}")
report(trades_c, f"C: 트레일링 10% + 타임스탑 24h | {label}")
report(trades_d, f"D: 트레일링 10% + 타임스탑 없음 | {label}")
def main_combo_cmp(interval: str = DEFAULT_INTERVAL) -> None:
"""추세 임계값 + 거래량 배수 조합 비교.
목적: 거래 빈도를 줄여 수수료 부담을 낮추는 최적 조합 탐색.
A (현행): trend=1.0%, vol=2x
B: trend=1.5%, vol=2x (추세만 강화)
C: trend=1.0%, vol=3x (거래량만 강화)
D: trend=1.5%, vol=3x (둘 다 강화)
E: trend=2.0%, vol=3x (최강 강화)
"""
cfg = INTERVAL_CONFIG[interval]
label = cfg["label"]
base_trend = cfg["trend"] # 15분봉 기준 1.0%
combos = [
("A 현행 1.0%/2x", base_trend, 2.0),
("B 추세1.5%/2x", base_trend * 1.5, 2.0),
("C 추세1.0%/3x", base_trend, 3.0),
("D 추세1.5%/3x", base_trend * 1.5, 3.0),
("E 추세2.0%/3x", base_trend * 2.0, 3.0),
]
print(f"\n{'='*64}")
print(f" 추세+거래량 조합 비교 | {MONTHS}개월 | 상위 {TOP_N}종목 | {label}")
print(f" 목적: 거래 빈도↓ → 수수료 부담↓ → 순수익↑")
print(f" FEE = 0.05% × 2 = 0.1%/건 | 현행 ~85건/월 = 월 8.5% 수수료")
print(f"{'='*64}")
print("\n▶ 종목 데이터 로드 중 (DB 캐시 우선)...")
tickers = get_top_tickers(TOP_N)
data = fetch_all_data(tickers, interval)
print(f" 사용 종목: {list(data.keys())}")
daily_features = {t: build_daily_features(f["daily"]) for t, f in data.items()}
results = []
for combo_label, trend, vol in combos:
print(f"\n▶ {combo_label} (trend={trend*100:.1f}%, vol={vol:.0f}x) ...")
trades = run_scenario(
data, stop_loss=0.015,
trend_min_gain=trend, volume_mult=vol,
reentry_above_sell=True,
interval_cd=interval, use_db_cache=True,
daily_features=daily_features,
)
n = len(trades)
fee_pct = n * 0.1 # 총 수수료 부담 (%)
print(f" 완료 ({n}건 | 예상 수수료 {fee_pct:.1f}%)")
results.append((combo_label, trades))
compare_report(results, title=f"추세+거래량 조합 비교 ({label})")
# 수수료 분석 추가 출력
print(" [수수료 분석]")
print(f" {'조합':<20} {'거래수':>6} {'월평균':>6} {'총수수료':>8} {'수수료 전':>10} {'수수료 후':>10}")
print(f" {'-'*64}")
for (clabel, trades) in results:
if not trades:
continue
df = pd.DataFrame(trades)
n = len(df)
span_days = (df["exit"].max() - df["entry"].min()).total_seconds() / 86400
per_month = n / (span_days / 30) if span_days > 0 else 0
fee_total = n * 0.1
net_pnl = df["pnl_pct"].sum()
gross_pnl = net_pnl + fee_total # 수수료 전 추정
print(f" {clabel:<20} {n:>6}건 {per_month:>5.0f}/월 {fee_total:>7.1f}% "
f"{gross_pnl:>+9.1f}% {net_pnl:>+9.1f}%")
print()
for clabel, trades in results:
report(trades, clabel)
def main() -> None:
if COMPARE_TOP50:
if COMPARE_COMBO:
main_combo_cmp(DEFAULT_INTERVAL)
elif COMPARE_TIMESTOP:
main_timestop_cmp(DEFAULT_INTERVAL)
elif COMPARE_TOP50:
main_top50_cmp(DEFAULT_INTERVAL)
elif COMPARE_HARDSTOP:
main_hard_stop_cmp(DEFAULT_INTERVAL)