[2026-05-27] 포맷 후 복구 설치 스크립트 추가

app/ml/__init__.py

@@ -0,0 +1 @@
+

app/ml/features.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from typing import Iterable
+
+import pandas as pd
+
+
+LABEL_COLUMNS = {"label_win", "label_stop_loss"}
+
+EXCLUDED_COLUMNS = {
+    "id",
+    "trade_id",
+    "date",
+    "ticker",
+    "name",
+    "entry_time",
+    "exit_time",
+    "sample_time",
+    "created_at",
+    "exit_reason",
+    "strategy",
+    "reason",
+    "source_file",
+    "ai_win_score",
+    "ai_stop_loss_score",
+    "ai_model_version",
+}
+
+
+def select_feature_columns(df: pd.DataFrame, targets: Iterable[str] = LABEL_COLUMNS) -> list[str]:
+    excluded = EXCLUDED_COLUMNS | set(targets)
+    numeric_columns = [
+        column
+        for column in df.columns
+        if column not in excluded and pd.api.types.is_numeric_dtype(df[column])
+    ]
+    return sorted(numeric_columns)
+
+
+def build_feature_matrix(
+    df: pd.DataFrame,
+    feature_columns: list[str],
+    medians: dict[str, float] | None = None,
+) -> tuple[pd.DataFrame, dict[str, float]]:
+    features = df.reindex(columns=feature_columns)
+    features = features.apply(pd.to_numeric, errors="coerce")
+
+    if medians is None:
+        medians = {
+            column: float(value) if pd.notna(value) else 0.0
+            for column, value in features.median(numeric_only=True).items()
+        }
+
+    features = features.fillna(medians).fillna(0.0)
+    return features, medians

app/ml/predictor.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import joblib
+import pandas as pd
+
+from app.ml.features import build_feature_matrix
+
+
+DEFAULT_MODEL_PATH = Path("models/scalping_model.joblib")
+
+
+class ScalpingModel:
+    def __init__(self, model_path: str | Path = DEFAULT_MODEL_PATH):
+        self.model_path = Path(model_path)
+        self.bundle: dict[str, Any] | None = None
+
+    @property
+    def available(self) -> bool:
+        return self.model_path.exists()
+
+    def load(self) -> bool:
+        if not self.available:
+            return False
+        self.bundle = joblib.load(self.model_path)
+        return True
+
+    @property
+    def version(self) -> str:
+        if self.bundle is None and not self.load():
+            return ""
+        assert self.bundle is not None
+        return str(self.bundle.get("created_at", ""))
+
+    def score(self, row: dict[str, Any]) -> dict[str, float]:
+        if self.bundle is None and not self.load():
+            return {}
+
+        assert self.bundle is not None
+        frame = pd.DataFrame([row])
+        features, _ = build_feature_matrix(
+            frame,
+            self.bundle["feature_columns"],
+            self.bundle.get("medians"),
+        )
+
+        scores: dict[str, float] = {}
+        for target, model in self.bundle.get("models", {}).items():
+            if hasattr(model, "predict_proba"):
+                scores[target] = float(model.predict_proba(features)[0][1])
+            else:
+                scores[target] = float(model.predict(features)[0])
+        return scores