"""
general_analysis 봉 데이터 전구간 enrich + 최신봉·기법 점검 리포트.
python scripts/03_analyze_enrich.py
python scripts/03_analyze_enrich.py --interval 1440
"""
from __future__ import annotations
import argparse
from pathlib import Path
import pandas as pd
from config import CHART_LOOKBACK_DAYS, GA_DEFAULT_TAIL_EXPORT, SYMBOL
from deepcoin.analysis.general_analysis_align import (
general_analysis_mtf_scores,
general_analysis_mtf_vote_latest,
)
from deepcoin.analysis.general_analysis_config import (
DEFAULT_CAPABILITY_HTML,
DEFAULT_LATEST_DIR,
GENERAL_ANALYSIS_INTERVALS,
)
from deepcoin.analysis.general_analysis_core import ga_col, interval_tf_prefix
from deepcoin.analysis.general_analysis_pipeline import general_analysis_enrich_bars
from deepcoin.ops.monitor import Monitor
from deepcoin.data.mtf_bb import load_frames_from_db
def _latest_row_summary(df: pd.DataFrame, prefix: str) -> dict[str, object]:
"""최신 봉의 ga_·핵심 레거시 컬럼 요약."""
if df.empty:
return {}
row = df.iloc[-1]
out: dict[str, object] = {"dt": str(df.index[-1]), "tf": prefix}
for c in df.columns:
if c.startswith("ga_") or c in ("RSI", "bb_pos", "macd_hist", "stoch_k"):
v = row[c]
if pd.isna(v):
continue
out[c] = v
return out
def write_capability_html(
summaries: dict[str, dict[str, object]],
vote: dict[str, object],
path: Path,
) -> None:
"""기법 컬럼 존재 여부·최신값 요약 HTML."""
rows = ""
for tf, snap in summaries.items():
ga_cols = [k for k in snap if str(k).startswith("ga_")]
rows += f"| {tf} | {snap.get('dt','')} | {len(ga_cols)} |
"
vote_rows = "".join(f"{k}: {v}" for k, v in vote.items())
html = f"""
general_analysis 기법 점검
general_analysis 기법 점검 ({SYMBOL})
3분~일봉 enrich 완료. 최신 봉 기준 컬럼 수·MTF 투표.
간격별 ga_ 컬럼 수
MTF 투표 (최신 봉)
상세 CSV: {DEFAULT_LATEST_DIR}/
"""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(html, encoding="utf-8")
def main() -> None:
parser = argparse.ArgumentParser(description="general_analysis 8TF enrich")
parser.add_argument("--interval", type=int, default=0, help="단일 간격만 (0=전체)")
parser.add_argument(
"--tail-export",
type=int,
default=GA_DEFAULT_TAIL_EXPORT,
help="CSV 저장 최근 N봉",
)
args = parser.parse_args()
from deepcoin.paths import ANALYSIS_CAPABILITY_HTML, ANALYSIS_LATEST_DIR
intervals = (
(args.interval,)
if args.interval > 0
else GENERAL_ANALYSIS_INTERVALS
)
print(f"=== general_analysis enrich {SYMBOL} ===")
mon = Monitor(cooldown_file=None)
frames = load_frames_from_db(mon, SYMBOL, lookback_days=CHART_LOOKBACK_DAYS)
if not frames:
raise RuntimeError("coins.db 데이터 없음")
enriched: dict[int, pd.DataFrame] = {}
summaries: dict[str, dict[str, object]] = {}
out_dir = ANALYSIS_LATEST_DIR
out_dir.mkdir(parents=True, exist_ok=True)
for iv in intervals:
raw = frames.get(iv)
if raw is None or raw.empty:
print(f" skip {iv}: no data")
continue
p = interval_tf_prefix(iv)
print(f" [{p}] enrich {len(raw)} bars...")
ef = general_analysis_enrich_bars(raw, iv, full_context=True)
enriched[iv] = ef
tail = ef.tail(args.tail_export)
csv_path = out_dir / f"{p}_latest.csv"
tail.to_csv(csv_path, encoding="utf-8-sig")
print(f" -> {csv_path} ({len(tail)} rows x {len(tail.columns)} cols)")
summaries[p] = _latest_row_summary(ef, p)
flat_vote: dict[str, object] = {}
if len(enriched) >= 2:
for k, v in general_analysis_mtf_vote_latest(enriched).items():
flat_vote[ga_col(k)] = v
prefixed = {}
for iv, ef in enriched.items():
p = interval_tf_prefix(iv)
row = ef.iloc[-1]
for c in ("RSI", "bb_pos"):
if c in row.index:
prefixed[f"{p}_{c}"] = row[c]
st = row.get(ga_col("struct_trend"))
if st is not None:
prefixed[f"{p}_{ga_col('struct_trend')}"] = st
flat_vote.update(general_analysis_mtf_scores(prefixed))
write_capability_html(summaries, flat_vote, ANALYSIS_CAPABILITY_HTML)
print(f"점검 리포트: {cap_path}")
print("완료.")
if __name__ == "__main__":
main()