dsyoon
b52d61b777
로고스/루트 레거시를 제거하고 deepcoin 패키지·scripts 01~05 CLI·docs/reference로 데이터·GT·분석·매칭·운영 단계를 정리했다. config와 .env 기반 설정, trade_anaysis.html 동기화 포함. Co-authored-by: Cursor <cursoragent@cursor.com>
93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
"""
|
|
general_analysis 공통 유틸 (슬라이스·피벗·컬럼 접두사).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from deepcoin.analysis.general_analysis_config import GA_COL_PREFIX, LOOKBACK_BARS
|
|
|
|
|
|
def ga_col(name: str) -> str:
|
|
"""general_analysis 출력 컬럼명."""
|
|
return f"{GA_COL_PREFIX}{name}"
|
|
|
|
|
|
def interval_tf_prefix(interval: int) -> str:
|
|
"""간격 접두사 (m3, d1)."""
|
|
from deepcoin.analysis.general_analysis_config import INTERVAL_PREFIX
|
|
|
|
return INTERVAL_PREFIX.get(interval, f"m{interval}")
|
|
|
|
|
|
def prefixed_snapshot(
|
|
row: pd.Series,
|
|
interval: int,
|
|
keys: list[str] | tuple[str, ...],
|
|
) -> dict[str, Any]:
|
|
"""한 봉의 ga_ 컬럼을 {m3_ga_rsi: ...} 형태로 변환."""
|
|
p = interval_tf_prefix(interval)
|
|
out: dict[str, Any] = {}
|
|
for k in keys:
|
|
col = ga_col(k)
|
|
if col in row.index:
|
|
v = row[col]
|
|
out[f"{p}_{col}"] = None if pd.isna(v) else v
|
|
return out
|
|
|
|
|
|
def slice_to_timestamp(df: pd.DataFrame, ts: pd.Timestamp) -> pd.DataFrame:
|
|
"""타점 시각 이전 완성봉만 (해당 시각 봉 미포함)."""
|
|
if df.empty:
|
|
return df
|
|
if not isinstance(df.index, pd.DatetimeIndex):
|
|
df = df.copy()
|
|
df.index = pd.to_datetime(df.index)
|
|
ts = pd.Timestamp(ts)
|
|
if ts.tzinfo is not None and df.index.tz is None:
|
|
ts = ts.tz_localize(None)
|
|
return df[df.index < ts].copy()
|
|
|
|
|
|
def lookback_slice(df: pd.DataFrame, interval: int, end_ts: pd.Timestamp) -> pd.DataFrame:
|
|
"""타점 직전 lookback 구간."""
|
|
sliced = slice_to_timestamp(df, end_ts)
|
|
n = LOOKBACK_BARS.get(interval, 80)
|
|
if len(sliced) > n:
|
|
return sliced.iloc[-n:].copy()
|
|
return sliced
|
|
|
|
|
|
def find_pivots(
|
|
highs: np.ndarray,
|
|
lows: np.ndarray,
|
|
order: int = 3,
|
|
) -> tuple[list[int], list[int]]:
|
|
"""국소 고점·저점 인덱스 (양쪽 order 봉보다 극값)."""
|
|
peak_idx: list[int] = []
|
|
trough_idx: list[int] = []
|
|
n = len(highs)
|
|
if n < order * 2 + 1:
|
|
return peak_idx, trough_idx
|
|
for i in range(order, n - order):
|
|
if highs[i] >= highs[i - order : i + order + 1].max():
|
|
peak_idx.append(i)
|
|
if lows[i] <= lows[i - order : i + order + 1].min():
|
|
trough_idx.append(i)
|
|
return peak_idx, trough_idx
|
|
|
|
|
|
def last_row_dict(df: pd.DataFrame, cols: list[str]) -> dict[str, Any]:
|
|
"""마지막 봉의 지정 컬럼 dict."""
|
|
if df.empty:
|
|
return {ga_col(c): None for c in cols}
|
|
row = df.iloc[-1]
|
|
return {
|
|
ga_col(c): (None if c not in row.index or pd.isna(row[c]) else row[c])
|
|
for c in cols
|
|
}
|