dsyoon/Bithumb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

로고스 전략 FSM을 simulation 기본 실행에 통합한다.

Browse Source 
수동 타점(logos_trades.json) 흐름에 맞춘 순차 매매 로직을 추가하고, python simulation.py 실행 시 로고스 백테스트·HTML을 생성한다. 규칙 탐색·BB 안전장치 개선과 함께 reports HTML은 gitignore로 제외한다.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dsyoon
2026-05-29 19:07:10 +09:00
parent e218a8ea32
commit e631a5701f
12 changed files with 1639 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

365
rule_discovery.py
View File

@@ -21,12 +21,18 @@ from candle_features import (
)
from config import (
BUY_COOLDOWN_SEC,
BUY_MAX_BB_POS_CHASE,
DISCOVER_MAX_TRADES,
DISCOVER_TRADE_PENALTY_PCT,
DOWNLOAD_INTERVALS,
ENTRY_INTERVAL,
SELL_COOLDOWN_SEC,
SELL_MIN_BB_POS,
SIGNAL_EDGE_ONLY,
SIM_INITIAL_CASH_KRW,
SIM_MIN_ORDER_KRW,
SYMBOL,
TRADE_MIN_GAP_BARS,
TRADING_FEE_RATE,
)
from strategy import (
@@ -59,6 +65,19 @@ BUY_SAFETY_BLOCK: tuple[str, ...] = (
"m10:cross_up_upper",
)
# 연속 봉에서 오래 참 → 엣지 없으면 과다 체결
LEVEL_STATE_FEATURES: tuple[str, ...] = (
"below_lower",
"above_upper",
"inside_band",
"bb_zone_bottom",
"bb_zone_top",
"bb_pos_low",
"bb_pos_high",
"ichi_price_above_tenkan",
"ichi_price_below_tenkan",
)
@dataclass
class DiscoveredRules:
@@ -87,6 +106,104 @@ def predicate_column(key: str) -> tuple[str, bool]:
return f"{prefix}_{feat}", neg
def _predicate_feature(key: str) -> str:
"""predicate에서 특징명만 추출 (! 제외)."""
rest = key.split(":", 1)[1]
return rest[1:] if rest.startswith("!") else rest
def is_level_state_predicate(key: str) -> bool:
"""한번 참이면 여러 봉 연속 참인 상태형 조건."""
return _predicate_feature(key) in LEVEL_STATE_FEATURES
def is_weak_sell_predicate(key: str) -> bool:
"""
!cross_* / !below_* 등 — 대부분의 봉에서 참이라 매도가 과다해짐.
"""
if ":" not in key:
return False
rest = key.split(":", 1)[1]
if not rest.startswith("!"):
return False
feat = rest[1:]
if feat.startswith("cross_"):
return True
return feat in ("below_lower", "above_upper", "inside_band")
def is_blocked_buy_predicate(key: str) -> bool:
"""진입(3분) 봉의 상태형 매수 조건은 탐색에서 제외."""
pfx = interval_prefix(ENTRY_INTERVAL)
return key.startswith(f"{pfx}:") and is_level_state_predicate(key)
# 고점 추격 매수(상단 구간·과열) — 탐색·체결에서 제외
CHASE_BUY_FEATURES: tuple[str, ...] = (
"bb_zone_top",
"bb_zone_high",
"bb_pos_high",
"above_upper",
"cross_up_upper",
)
# 저점·반등 매수 트리거
VALUE_BUY_FEATURES: tuple[str, ...] = (
"cross_up_lower",
"bb_zone_bottom",
"bb_zone_low",
"hammer",
"bb_pos_low",
"ichi_tk_cross_up",
"cross_down_lower",
)
def is_chase_buy_predicate(key: str) -> bool:
"""밴드 상단·고점 추격 매수 조건."""
if ":" not in key:
return False
rest = key.split(":", 1)[1]
if rest.startswith("!"):
return False
return _predicate_feature(key) in CHASE_BUY_FEATURES
def is_value_buy_predicate(key: str) -> bool:
"""하단 돌파·반등형 매수 조건."""
if ":" not in key:
return False
rest = key.split(":", 1)[1]
if rest.startswith("!"):
return False
return _predicate_feature(key) in VALUE_BUY_FEATURES
def _entry_bb_pos_col() -> str:
return f"{interval_prefix(ENTRY_INTERVAL)}_bb_pos"
def discover_score(return_pct: float, trade_count: int) -> float:
"""탐색 목적함수: 수익률 − 과다 거래 패널티."""
excess = max(0, trade_count - DISCOVER_MAX_TRADES)
return return_pct - excess * DISCOVER_TRADE_PENALTY_PCT
def _rising_edge(mask: np.ndarray, i: int) -> bool:
"""i번째 봉에서 조건이 새로 참이 됐는지."""
if not bool(mask[i]):
return False
if i == 0:
return True
return not bool(mask[i - 1])
def _trigger_at(mask: np.ndarray, i: int, edge_only: bool = SIGNAL_EDGE_ONLY) -> bool:
if edge_only:
return _rising_edge(mask, i)
return bool(mask[i])
def _mask_for_keys(matrix: pd.DataFrame, keys: list[str]) -> np.ndarray:
"""AND 조건 마스크."""
n = len(matrix)
@@ -134,9 +251,59 @@ def _unsafe_buy_mask(matrix: pd.DataFrame) -> np.ndarray:
unsafe |= (
matrix["m30_hammer"].fillna(0).astype(bool) & near_peak.fillna(False)
).to_numpy()
if _entry_bb_pos_col() in matrix.columns:
pos = matrix[_entry_bb_pos_col()].fillna(0.5).astype(float).to_numpy()
unsafe |= pos >= BUY_MAX_BB_POS_CHASE
for key in CHASE_BUY_FEATURES:
col = f"{interval_prefix(ENTRY_INTERVAL)}_{key}"
if col in matrix.columns:
unsafe |= matrix[col].fillna(0).astype(bool).to_numpy()
return unsafe
def _value_buy_gate_mask(matrix: pd.DataFrame, group: list[str]) -> np.ndarray:
"""
매수 그룹별: 저점 트리거(value) 또는 3분 bb_pos < BUY_MAX_BB_POS_CHASE 일 때만 허용.
"""
n = len(matrix)
pos_col = _entry_bb_pos_col()
if pos_col in matrix.columns:
pos_ok = (
matrix[pos_col].fillna(0.5).astype(float).to_numpy()
< BUY_MAX_BB_POS_CHASE
)
else:
pos_ok = np.ones(n, dtype=bool)
value_keys = [k for k in group if is_value_buy_predicate(k)]
if not value_keys:
return pos_ok
value_hit = _mask_for_keys(matrix, value_keys)
return pos_ok | value_hit
def _unsafe_sell_mask(matrix: pd.DataFrame) -> np.ndarray:
"""
저점·반등 구간 매도 차단.
- 3분 bb_pos < SELL_MIN_BB_POS
- 망치·밴드 하단 구간에서 상단돌파 익절 방지 (5/26 01:48 유형)
"""
n = len(matrix)
blocked = np.zeros(n, dtype=bool)
pos_col = _entry_bb_pos_col()
if pos_col in matrix.columns:
pos = matrix[pos_col].fillna(0.5).astype(float).to_numpy()
blocked |= pos < SELL_MIN_BB_POS
pfx = interval_prefix(ENTRY_INTERVAL)
for feat in ("hammer", "bb_zone_bottom", "bb_zone_low", "bb_pos_low"):
col = f"{pfx}_{feat}"
if col in matrix.columns:
blocked |= matrix[col].fillna(0).astype(bool).to_numpy()
return blocked
def buy_mask(matrix: pd.DataFrame, rules: DiscoveredRules) -> np.ndarray:
"""
매수 마스크 = (buy_all) 또는 (buy_any 각 그룹의 AND) 중 하나 + 안전필터.
@@ -154,13 +321,34 @@ def buy_mask(matrix: pd.DataFrame, rules: DiscoveredRules) -> np.ndarray:
return np.zeros(n, dtype=bool)
any_ok = np.zeros(n, dtype=bool)
for group in groups:
any_ok |= _mask_for_keys(matrix, group)
raw = _mask_for_keys(matrix, group)
any_ok |= raw & _value_buy_gate_mask(matrix, group)
return any_ok & ~_unsafe_buy_mask(matrix)
def sell_mask(matrix: pd.DataFrame, rules: DiscoveredRules, stop: bool = False) -> np.ndarray:
keys = rules.sell_stop if stop else rules.sell_all
return _mask_for_keys(matrix, keys)
raw = _mask_for_keys(matrix, keys)
if stop:
return raw
return raw & ~_unsafe_sell_mask(matrix)
def sanitize_rules(rules: DiscoveredRules) -> DiscoveredRules:
"""탐색 결과에서 추격 매수·무의미 조건 제거."""
rules.buy_all = [p for p in rules.buy_all if not is_chase_buy_predicate(p)]
rules.buy_any = [
[p for p in g if not is_chase_buy_predicate(p)]
for g in rules.buy_any
]
rules.buy_any = [g for g in rules.buy_any if g]
rules.sell_all = [p for p in rules.sell_all if not is_weak_sell_predicate(p)]
return rules
def _discovery_seed() -> DiscoveredRules:
"""탐색 시드: 하단 돌파 기준선 (combination_seed의 상단 추격 매수 미사용)."""
return _baseline_rules()
def generate_predicate_pool(intervals: list[int]) -> list[str]:
@@ -176,6 +364,35 @@ def generate_predicate_pool(intervals: list[int]) -> list[str]:
return pool
def list_rule_signal_edges(
matrix: pd.DataFrame,
rules: DiscoveredRules,
) -> list[tuple[pd.Timestamp, str]]:
"""
전 기간 규칙 엣지 신호(체결 여부와 무관).
Returns:
(timestamp, action) — buy_signal | sell_signal | sell_stop_signal
"""
idx = matrix.index
b_mask = buy_mask(matrix, rules)
s_mask = sell_mask(matrix, rules, stop=False)
stop_mask = (
sell_mask(matrix, rules, stop=True)
if rules.sell_stop
else np.zeros(len(matrix), dtype=bool)
)
out: list[tuple[pd.Timestamp, str]] = []
for i in range(len(matrix)):
if _rising_edge(b_mask, i):
out.append((idx[i], "buy_signal"))
if _rising_edge(s_mask, i):
out.append((idx[i], "sell_signal"))
if rules.sell_stop and _rising_edge(stop_mask, i):
out.append((idx[i], "sell_stop_signal"))
return out
def generate_trade_events(
matrix: pd.DataFrame,
rules: DiscoveredRules,
@@ -200,6 +417,7 @@ def generate_trade_events(
qty = 0.0
last_buy_i: int | None = None
last_sell_i: int | None = None
last_trade_i: int | None = None
for i in range(len(matrix)):
price = close[i]
@@ -207,9 +425,12 @@ def generate_trade_events(
continue
ts = idx[i]
if last_trade_i is not None and i - last_trade_i < TRADE_MIN_GAP_BARS:
continue
if qty > 0:
is_stop = bool(stop_mask[i])
is_sell = bool(s_mask[i])
is_stop = _trigger_at(stop_mask, i) if rules.sell_stop else False
is_sell = _trigger_at(s_mask, i)
if is_stop or is_sell:
if last_sell_i is not None:
if (ts - idx[last_sell_i]).total_seconds() < SELL_COOLDOWN_SEC:
@@ -218,15 +439,17 @@ def generate_trade_events(
events.append((ts, "sell", sig))
qty = 0.0
last_sell_i = i
last_trade_i = i
continue
if b_mask[i] and qty <= 0:
if _trigger_at(b_mask, i) and qty <= 0:
if last_buy_i is not None:
if (ts - idx[last_buy_i]).total_seconds() < BUY_COOLDOWN_SEC:
continue
events.append((ts, "buy", SIGNAL_BUY_LOWER))
qty = 1.0
last_buy_i = i
last_trade_i = i
return events
@@ -263,6 +486,18 @@ def backtest_rules(
return res.total_return_pct, res.trade_count
def _evaluate_train(
train: pd.DataFrame,
rules: DiscoveredRules,
df_1d: pd.DataFrame,
df_1h: pd.DataFrame,
entry_ohlc: pd.DataFrame,
) -> tuple[float, int, float]:
"""학습 구간 수익·거래수·목적함수 점수."""
ret, tc = backtest_rules(train, rules, df_1d, df_1h, entry_ohlc)
return ret, tc, discover_score(ret, tc)
def _baseline_rules() -> DiscoveredRules:
"""다봉 BB 하단 돌파 + 상단 돌파 기준선."""
p3 = interval_prefix(ENTRY_INTERVAL)
@@ -321,13 +556,22 @@ def greedy_search(
sell_all=list(seed.sell_all),
sell_stop=list(seed.sell_stop),
)
best_ret, _ = backtest_rules(train, best, df_1d, df_1h, entry_ohlc)
best_ret, best_tc, best_score = _evaluate_train(
train, best, df_1d, df_1h, entry_ohlc
)
buy_pool = [
p
for p in pool
if not is_blocked_buy_predicate(p) and not is_chase_buy_predicate(p)
]
sell_pool = [p for p in pool if not is_weak_sell_predicate(p)]
improved = True
while improved:
improved = False
# 매수 AND 추가/제거
for pred in pool:
for pred in buy_pool:
if pred in best.buy_all:
trial_all = [p for p in best.buy_all if p != pred]
else:
@@ -341,14 +585,16 @@ def greedy_search(
sell_all=best.sell_all,
sell_stop=best.sell_stop,
)
ret, _ = backtest_rules(train, trial, df_1d, df_1h, entry_ohlc)
if ret > best_ret:
best_ret = ret
ret, tc, score = _evaluate_train(
train, trial, df_1d, df_1h, entry_ohlc
)
if score > best_score:
best_ret, best_tc, best_score = ret, tc, score
best.buy_all = trial_all
improved = True
# 매도 AND
for pred in pool:
for pred in sell_pool:
if pred in best.sell_all:
trial_s = [p for p in best.sell_all if p != pred]
else:
@@ -362,14 +608,21 @@ def greedy_search(
sell_all=trial_s,
sell_stop=best.sell_stop,
)
ret, _ = backtest_rules(train, trial, df_1d, df_1h, entry_ohlc)
if ret > best_ret:
best_ret = ret
ret, tc, score = _evaluate_train(
train, trial, df_1d, df_1h, entry_ohlc
)
if score > best_score:
best_ret, best_tc, best_score = ret, tc, score
best.sell_all = trial_s
improved = True
# 손절
stop_pool = [p for p in pool if "cross_down_lower" in p or "below_lower" in p]
stop_pool = [
p
for p in pool
if "cross_down_lower" in p
and not is_level_state_predicate(p)
]
for pred in stop_pool:
if pred in best.sell_stop:
trial_st = [p for p in best.sell_stop if p != pred]
@@ -384,13 +637,15 @@ def greedy_search(
sell_all=best.sell_all,
sell_stop=trial_st,
)
ret, _ = backtest_rules(train, trial, df_1d, df_1h, entry_ohlc)
if ret > best_ret:
best_ret = ret
ret, tc, score = _evaluate_train(
train, trial, df_1d, df_1h, entry_ohlc
)
if score > best_score:
best_ret, best_tc, best_score = ret, tc, score
best.sell_stop = trial_st
improved = True
return best
return sanitize_rules(best)
def try_buy_any_branches(
@@ -420,7 +675,9 @@ def try_buy_any_branches(
sell_all=list(base.sell_all),
sell_stop=list(base.sell_stop),
)
best_ret, _ = backtest_rules(train, best, df_1d, df_1h, entry_ohlc)
best_ret, best_tc, best_score = _evaluate_train(
train, best, df_1d, df_1h, entry_ohlc
)
for pred in triggers[:max_branches]:
if pred in best.buy_all:
@@ -434,13 +691,15 @@ def try_buy_any_branches(
)
if not trial.buy_any[0]:
trial.buy_any = [[pred]]
ret, _ = backtest_rules(train, trial, df_1d, df_1h, entry_ohlc)
if ret > best_ret:
best_ret = ret
best = trial
ret, tc, score = _evaluate_train(
train, trial, df_1d, df_1h, entry_ohlc
)
if score > best_score:
best_ret, best_score = ret, score
best = sanitize_rules(trial)
best.name = "discovered_or"
return best
return sanitize_rules(best)
def random_search_refine(
@@ -456,8 +715,16 @@ def random_search_refine(
"""무작위 변형으로 국소 최적 보완."""
train = matrix.iloc[:train_end]
best = seed
best_ret, _ = backtest_rules(train, best, df_1d, df_1h, entry_ohlc)
best_ret, best_tc, best_score = _evaluate_train(
train, best, df_1d, df_1h, entry_ohlc
)
rng = random.Random(42)
buy_pool = [
p
for p in pool
if not is_blocked_buy_predicate(p) and not is_chase_buy_predicate(p)
]
sell_pool = [p for p in pool if not is_weak_sell_predicate(p)]
for _ in range(iterations):
trial = DiscoveredRules(
@@ -468,27 +735,30 @@ def random_search_refine(
sell_stop=[p for p in best.sell_stop],
)
action = rng.choice(["add_buy", "drop_buy", "add_sell", "drop_sell", "swap_buy"])
if action == "add_buy" and len(trial.buy_all) < 6:
p = rng.choice(pool)
if action == "add_buy" and len(trial.buy_all) < 6 and buy_pool:
p = rng.choice(buy_pool)
if p not in trial.buy_all:
trial.buy_all.append(p)
elif action == "drop_buy" and trial.buy_all:
trial.buy_all.pop(rng.randrange(len(trial.buy_all)))
elif action == "add_sell" and len(trial.sell_all) < 5:
p = rng.choice(pool)
elif action == "add_sell" and len(trial.sell_all) < 5 and sell_pool:
p = rng.choice(sell_pool)
if p not in trial.sell_all:
trial.sell_all.append(p)
elif action == "drop_sell" and trial.sell_all:
trial.sell_all.pop(rng.randrange(len(trial.sell_all)))
elif action == "swap_buy" and pool:
elif action == "swap_buy" and buy_pool:
if trial.buy_all:
trial.buy_all[rng.randrange(len(trial.buy_all))] = rng.choice(pool)
ret, _ = backtest_rules(train, trial, df_1d, df_1h, entry_ohlc)
if ret > best_ret:
best_ret = ret
trial.buy_all[rng.randrange(len(trial.buy_all))] = rng.choice(buy_pool)
trial = sanitize_rules(trial)
ret, tc, score = _evaluate_train(
train, trial, df_1d, df_1h, entry_ohlc
)
if score > best_score:
best_ret, best_score = ret, score
best = trial
best.name = "discovered_refined"
return best
return sanitize_rules(best)
def discover_rules(frames: dict[int, pd.DataFrame]) -> DiscoveredRules:
@@ -513,11 +783,13 @@ def discover_rules(frames: dict[int, pd.DataFrame]) -> DiscoveredRules:
pool = generate_predicate_pool(intervals)
print(f" 샘플 {n}봉 | 학습 {train_end} | predicate 후보 {len(pool)}개")
baseline = _seed_from_combination_report() or _baseline_rules()
br, bt = backtest_rules(matrix.iloc[:train_end], baseline, df_1d, df_1h, entry_ohlc)
print(f" 시드 규칙: {baseline.name}")
baseline = _discovery_seed()
br, bt = backtest_rules(
matrix.iloc[:train_end], baseline, df_1d, df_1h, entry_ohlc
)
print(f" 시드 규칙: {baseline.name} (하단돌파 매수·상단돌파 매도)")
bf, _ = backtest_rules(matrix, baseline, df_1d, df_1h, entry_ohlc)
print(f" 기준선(3분 BB만): 학습 {br:+.2f}% | 전체 {bf:+.2f}%")
print(f" 기준선: 학습 {br:+.2f}% | 전체 {bf:+.2f}%")
print("1단계: 탐욕적 AND 확장...")
g1 = greedy_search(matrix, train_end, pool, baseline, df_1d, df_1h, entry_ohlc)
@@ -532,8 +804,13 @@ def discover_rules(frames: dict[int, pd.DataFrame]) -> DiscoveredRules:
print("3단계: 무작위 정밀 탐색...")
best = g2 if r2 >= r1 else g1
g3 = random_search_refine(matrix, train_end, pool, best, df_1d, df_1h, entry_ohlc, iterations=1200)
train_ret, t_cnt = backtest_rules(matrix.iloc[:train_end], g3, df_1d, df_1h, entry_ohlc)
test_ret, _ = backtest_rules(matrix.iloc[train_end:], g3, df_1d, df_1h, entry_ohlc)
g3 = sanitize_rules(g3)
train_ret, t_cnt = backtest_rules(
matrix.iloc[:train_end], g3, df_1d, df_1h, entry_ohlc
)
test_ret, _ = backtest_rules(
matrix.iloc[train_end:], g3, df_1d, df_1h, entry_ohlc
)
full_ret, full_cnt = backtest_rules(matrix, g3, df_1d, df_1h, entry_ohlc)
g3.train_return_pct = train_ret
@@ -549,7 +826,9 @@ def discover_rules(frames: dict[int, pd.DataFrame]) -> DiscoveredRules:
print(f" 매도 AND: {g3.sell_all}")
if g3.sell_stop:
print(f" 손절: {g3.sell_stop}")
print(f" 학습 {train_ret:+.2f}% | 검증 {test_ret:+.2f}% | 전체 {full_ret:+.2f}% ({full_cnt}건)")
print(
f" 학습 {train_ret:+.2f}% | 검증 {test_ret:+.2f}% | 전체 {full_ret:+.2f}% ({full_cnt}건)"
)
return g3