ローソク足1本 = {
  open:   その時間足の最初の取引価格
  high:   その時間足の最高価格
  low:    その時間足の最安価格
  close:  その時間足の最後の取引価格
  volume: その時間足の総取引量
}

import yfinance as yf
import pandas as pd

def fetch_ohlcv(ticker: str, start: str, end: str, interval: str = "1d") -> pd.DataFrame:
    """
    Yahoo Finance から OHLCV データを取得する
    ticker  : ティッカーシンボル（例: "BTC-USD"）
    start   : 開始日（"YYYY-MM-DD"）
    end     : 終了日（"YYYY-MM-DD"）
    interval: 時間足（"1d", "1h", "15m" など）
    """
    df = yf.download(ticker, start=start, end=end, interval=interval, progress=False)
    df.columns = ["open", "high", "low", "close", "volume"]  # カラムを小文字統一
    df.index.name = "datetime"
    return df

# 取得例：BTCの日足データ（2021〜2025年）
btc = fetch_ohlcv("BTC-USD", start="2021-01-01", end="2025-12-31")

# 取得確認
print(f"データ件数   : {len(btc)} 行")
print(f"期間         : {btc.index[0]} 〜 {btc.index[-1]}")
print(f"欠損値の合計 : {btc.isnull().sum().sum()} 件")
print(btc.head())

def handle_missing(df: pd.DataFrame, gap_threshold_hours: int = 24) -> pd.DataFrame:
    """
    欠損値を補完する。長期欠損（gap_threshold_hours 超）は警告を出す。
    """
    # 欠損の連続長を確認
    missing_mask = df["close"].isnull()
    if missing_mask.any():
        # 連続欠損の最大幅を計算
        max_gap = missing_mask.astype(int).groupby(
            missing_mask.ne(missing_mask.shift()).cumsum()
        ).sum().max()
        if max_gap > gap_threshold_hours:
            print(f"[警告] {max_gap} 時間超の連続欠損を検出。該当期間の削除を推奨します。")

    # 短期欠損は前方補完（前の値で埋める）
    df = df.ffill()
    # 残った欠損（先頭部分）は後方補完
    df = df.bfill()
    return df

import numpy as np

def detect_outliers(df: pd.DataFrame, z_threshold: float = 3.0) -> pd.DataFrame:
    """
    Zスコアで外れ値を検出し、フラグ列を追加する
    """
    log_returns = np.log(df["close"] / df["close"].shift(1))
    z_scores = (log_returns - log_returns.mean()) / log_returns.std()

    df["outlier_flag"] = (z_scores.abs() > z_threshold).astype(int)

    # 価格変動20%超もフラグ
    price_change = df["close"].pct_change().abs()
    df["outlier_flag"] = df["outlier_flag"] | (price_change > 0.20).astype(int)

    print(f"外れ値検出数: {df['outlier_flag'].sum()} 件 ({df['outlier_flag'].mean():.2%})")
    return df

def add_log_return(df: pd.DataFrame) -> pd.DataFrame:
    """対数収益率を計算して列に追加する"""
    df["log_return"] = np.log(df["close"] / df["close"].shift(1))
    df.dropna(subset=["log_return"], inplace=True)
    return df

from sklearn.preprocessing import StandardScaler

# ⚠️ 重要：スケーラーは訓練データのみで fit する
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)

# テストデータには訓練データで fit したスケーラーを適用する（fit は不要）
X_test_scaled  = scaler.transform(X_test)

from imblearn.over_sampling import SMOTE

smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X_train, y_train)

print(f"リサンプリング前: {dict(zip(*np.unique(y_train, return_counts=True)))}")
print(f"リサンプリング後: {dict(zip(*np.unique(y_resampled, return_counts=True)))}")

def split_time_series(df: pd.DataFrame, train_ratio=0.8, val_ratio=0.1):
    """時系列データを順序を保って分割する"""
    n = len(df)
    train_end = int(n * train_ratio)
    val_end   = int(n * (train_ratio + val_ratio))

    train = df.iloc[:train_end]
    val   = df.iloc[train_end:val_end]
    test  = df.iloc[val_end:]

    print(f"訓練: {len(train)} 件 / 検証: {len(val)} 件 / テスト: {len(test)} 件")
    return train, val, test

def calc_rsi(df: pd.DataFrame, period: int = 14) -> pd.DataFrame:
    delta = df["close"].diff()
    gain  = delta.clip(lower=0)
    loss  = -delta.clip(upper=0)

    avg_gain = gain.ewm(com=period - 1, min_periods=period).mean()
    avg_loss = loss.ewm(com=period - 1, min_periods=period).mean()

    rs  = avg_gain / avg_loss
    df[f"rsi_{period}"] = 100 - (100 / (1 + rs))
    return df

def calc_macd(df: pd.DataFrame, fast=12, slow=26, signal=9) -> pd.DataFrame:
    ema_fast   = df["close"].ewm(span=fast, adjust=False).mean()
    ema_slow   = df["close"].ewm(span=slow, adjust=False).mean()

    df["macd"]        = ema_fast - ema_slow
    df["macd_signal"] = df["macd"].ewm(span=signal, adjust=False).mean()
    df["macd_hist"]   = df["macd"] - df["macd_signal"]
    return df

def calc_bollinger(df: pd.DataFrame, period: int = 20, num_std: float = 2.0) -> pd.DataFrame:
    sma   = df["close"].rolling(window=period).mean()
    std   = df["close"].rolling(window=period).std()

    df["bb_upper"]  = sma + num_std * std
    df["bb_lower"]  = sma - num_std * std
    df["bb_middle"] = sma

    # %B：現在価格がバンド内のどの位置にあるか（0〜1）
    df["bb_pct_b"]  = (df["close"] - df["bb_lower"]) / (df["bb_upper"] - df["bb_lower"])

    # バンド幅：ボラティリティの代理指標
    df["bb_width"]  = (df["bb_upper"] - df["bb_lower"]) / sma
    return df

def calc_ema(df: pd.DataFrame, periods: list = [9, 21, 50]) -> pd.DataFrame:
    for p in periods:
        df[f"ema_{p}"] = df["close"].ewm(span=p, adjust=False).mean()

    # クロスオーバーシグナル（短期EMAが長期EMAを上抜け）
    df["ema_cross_9_21"]  = (df["ema_9"] > df["ema_21"]).astype(int)
    df["ema_cross_21_50"] = (df["ema_21"] > df["ema_50"]).astype(int)
    return df

def add_lag_features(df: pd.DataFrame, target_col: str = "close",
                     lags: list = [1, 3, 7]) -> pd.DataFrame:
    for lag in lags:
        df[f"{target_col}_lag_{lag}"] = df[target_col].shift(lag)
    df.dropna(inplace=True)
    return df

def add_correlation_features(dfs: dict, window: int = 30) -> pd.DataFrame:
    """
    dfs: {"BTC": df_btc, "ETH": df_eth, "XRP": df_xrp} のような辞書
    各銘柄の終値を結合し、ローリング相関係数を計算する
    """
    closes = pd.DataFrame({k: v["close"] for k, v in dfs.items()})
    corr_btc_eth = closes["BTC"].rolling(window).corr(closes["ETH"])
    corr_btc_xrp = closes["BTC"].rolling(window).corr(closes["XRP"])

    dfs["BTC"]["corr_btc_eth"] = corr_btc_eth
    dfs["BTC"]["corr_btc_xrp"] = corr_btc_xrp
    return dfs["BTC"]

from sklearn.ensemble import RandomForestClassifier
import matplotlib.pyplot as plt

model_rf = RandomForestClassifier(n_estimators=100, random_state=42)
model_rf.fit(X_train, y_train)

importances = pd.Series(model_rf.feature_importances_, index=X_train.columns)
importances.sort_values(ascending=False).head(15).plot(kind="bar", figsize=(12, 5))
plt.title("特徴量の重要度 Top 15")
plt.tight_layout()
plt.show()

# 重要度が0.01未満の特徴量を除外
low_importance = importances[importances < 0.01].index.tolist()
print(f"除外する特徴量: {low_importance}")
X_train_selected = X_train.drop(columns=low_importance)

import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler

class CryptoPipeline:
    def __init__(self, ticker: str = "BTC-USD", random_seed: int = 42):
        self.ticker = ticker
        self.random_seed = random_seed
        np.random.seed(random_seed)
        self.scaler = StandardScaler()

    def run(self, df: pd.DataFrame) -> dict:
        df = handle_missing(df)
        df = detect_outliers(df)
        df = add_log_return(df)
        df = calc_rsi(df)
        df = calc_macd(df)
        df = calc_bollinger(df)
        df = calc_ema(df)
        df = add_lag_features(df)
        df.dropna(inplace=True)

        feature_cols = [c for c in df.columns if c not in ["open", "high", "low", "close", "volume"]]
        X = df[feature_cols]
        y = (df["close"].shift(-1) > df["close"]).astype(int)  # 翌日上昇=1

        train, val, test = split_time_series(df)
        X_train = train[feature_cols]
        X_val   = val[feature_cols]
        X_test  = test[feature_cols]

        X_train_scaled = self.scaler.fit_transform(X_train)
        X_val_scaled   = self.scaler.transform(X_val)
        X_test_scaled  = self.scaler.transform(X_test)

        return {
            "X_train": X_train_scaled, "X_val": X_val_scaled, "X_test": X_test_scaled,
            "y_train": train["log_return"].shift(-1).dropna(),
            "scaler": self.scaler,
            "feature_names": feature_cols,
        }

❌ 間違った手順：
全データ → fit_transform → 訓練/検証/テストに分割

✅ 正しい手順：
全データを分割 → 訓練データのみで fit → 各データセットに transform