import pandas as pd

df = pd.DataFrame({'close': [100, 105, 102, 108, 115]})

# ❌ NG：shift(0) → 当日終値でシグナルを出す → 未来情報が混入
df['signal_bad'] = df['close'].rolling(3).mean()

# ✅ OK：1期分シフトして未来情報を排除
df['signal_good'] = df['close'].rolling(3).mean().shift(1)
# shift(1)で「前日までの情報だけ」でシグナルを計算する

from sklearn.preprocessing import StandardScaler
import numpy as np

X = np.random.randn(1000, 5)

# ❌ NG：全データで fit_transform → テスト情報が学習に漏洩してしまう
# scaler.fit_transform(X)  ← これをやると未来情報が混入する

# ✅ OK：先にデータを分割してから、訓練データだけでfit
train_X = X[:800]
test_X  = X[800:]

scaler = StandardScaler()
train_X_scaled = scaler.fit_transform(train_X)  # 訓練データだけでfit
test_X_scaled  = scaler.transform(test_X)        # テストはtransformのみ

from bayes_opt import BayesianOptimization

def backtest_score(fast_period, slow_period):
    """バックテストのスコアを返す関数（シャープレシオなど）"""
    fast_period = int(fast_period)
    slow_period = int(slow_period)
    if fast_period >= slow_period:
        return -999
    sharpe = run_backtest(fast_period, slow_period)
    return sharpe

pbounds = {
    'fast_period': (5, 50),
    'slow_period': (20, 200),
}

optimizer = BayesianOptimization(
    f=backtest_score,
    pbounds=pbounds,
    random_state=42,
)

optimizer.maximize(init_points=5, n_iter=20)  # 試行回数を25回に抑制
print(optimizer.max)

import numpy as np

commission_rate = 0.001    # 手数料 0.1%
slippage_rate   = 0.003    # スリッページ 0.3%（流動性が低い場合）
trades_per_day  = 10
days            = 365

total_cost_per_trade = commission_rate + slippage_rate  # 0.4%/トレード

# 年間の総コスト影響（複利で計算）
cost_multiplier   = (1 - total_cost_per_trade) ** (trades_per_day * days)
total_cost_impact = 1 - cost_multiplier

print(f"年間コスト影響: {total_cost_impact:.2%}")
# 出力: 年間コスト影響: 76.89%

import pandas as pd
import numpy as np

# 時系列データの正しい分割方法
df = pd.read_csv('btc_ohlcv.csv', parse_dates=['timestamp'], index_col='timestamp')
df = df.sort_index()  # 時系列順にソート必須

# 時系列では shuffle=False が鉄則
split_point = int(len(df) * 0.8)

train_df = df.iloc[:split_point]    # 前80%：学習用
test_df  = df.iloc[split_point:]    # 後20%：テスト用（未来データ）

print(f"学習期間: {train_df.index[0]} ~ {train_df.index[-1]}")
print(f"テスト期間: {test_df.index[0]} ~ {test_df.index[-1]}")

import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import accuracy_score
from dateutil.relativedelta import relativedelta

def walk_forward_validation(df, train_months=18, test_months=1):
    """
    ウォークフォワード検証の実装

    Parameters:
    -----------
    df           : DataFrame（時系列の特徴量 + ターゲット変数）
    train_months : 学習窓の月数（デフォルト18ヶ月）
    test_months  : 検証窓の月数（デフォルト1ヶ月）
    """
    df = df.copy()
    df.index = pd.to_datetime(df.index)

    results = []
    start_date = df.index[0]
    end_date   = df.index[-1]

    current_train_end = start_date + relativedelta(months=train_months)

    while current_train_end + relativedelta(months=test_months) <= end_date:
        train_mask = (df.index >= start_date) & (df.index < current_train_end)
        test_start = current_train_end
        test_end   = current_train_end + relativedelta(months=test_months)
        test_mask  = (df.index >= test_start) & (df.index < test_end)

        train_data = df[train_mask]
        test_data  = df[test_mask]

        if len(train_data) < 100 or len(test_data) < 10:
            current_train_end += relativedelta(months=test_months)
            continue

        feature_cols = [c for c in df.columns if c != 'target']
        X_train = train_data[feature_cols]
        y_train = train_data['target']
        X_test  = test_data[feature_cols]
        y_test  = test_data['target']

        # モデルの再学習（毎月パラメータを更新）
        model = GradientBoostingClassifier(
            n_estimators=100,
            max_depth=4,
            learning_rate=0.05,
            random_state=42
        )
        model.fit(X_train, y_train)

        y_pred = model.predict(X_test)
        acc    = accuracy_score(y_test, y_pred)

        results.append({
            'train_end':      current_train_end,
            'test_period':    f"{test_start.strftime('%Y-%m')} ~ {test_end.strftime('%Y-%m')}",
            'accuracy':       acc,
            'n_test_samples': len(test_data),
        })

        print(f"テスト期間: {test_start.strftime('%Y-%m')} | Accuracy: {acc:.4f}")

        current_train_end += relativedelta(months=test_months)

    results_df = pd.DataFrame(results)
    print(f"\n平均Accuracy: {results_df['accuracy'].mean():.4f}")
    print(f"標準偏差:     {results_df['accuracy'].std():.4f}")
    return results_df

# 使い方
# results = walk_forward_validation(df, train_months=18, test_months=1)

import numpy as np
import pandas as pd

def calculate_backtest_metrics(returns: pd.Series) -> dict:
    """
    バックテスト評価指標の一括計算

    Parameters:
    -----------
    returns : pd.Series（日次リターン）
    """
    # シャープレシオ（年率換算）
    sharpe = returns.mean() / returns.std() * np.sqrt(252)

    # 最大ドローダウン
    cumulative   = (1 + returns).cumprod()
    rolling_max  = cumulative.cummax()
    drawdown     = (cumulative - rolling_max) / rolling_max
    max_drawdown = drawdown.min()

    # プロフィットファクター
    gross_profit  = returns[returns > 0].sum()
    gross_loss    = abs(returns[returns < 0].sum())
    profit_factor = gross_profit / gross_loss if gross_loss != 0 else np.inf

    # 勝率
    win_rate = (returns > 0).mean()

    return {
        'sharpe_ratio':  round(sharpe, 4),
        'max_drawdown':  round(max_drawdown, 4),
        'profit_factor': round(profit_factor, 4),
        'win_rate':      round(win_rate, 4),
        'total_return':  round((1 + returns).prod() - 1, 4),
        'n_trades':      len(returns),
    }

# 使い方
# returns = pd.Series([0.01, -0.005, 0.02, ...])
# metrics = calculate_backtest_metrics(returns)
# print(metrics)

import numpy as np
import pandas as pd

def monte_carlo_simulation(
    returns: pd.Series,
    n_simulations: int = 1000,
    n_days: int = 252
) -> pd.DataFrame:
    """
    モンテカルロシミュレーションによるリスク評価

    Parameters:
    -----------
    returns       : pd.Series（実績の日次リターン）
    n_simulations : シミュレーション回数（デフォルト1000回）
    n_days        : シミュレーション日数（デフォルト252日）
    """
    mu    = returns.mean()
    sigma = returns.std()

    simulation_results = []

    for _ in range(n_simulations):
        daily_returns      = np.random.normal(mu, sigma, n_days)
        cumulative_returns = (1 + daily_returns).cumprod()
        simulation_results.append(cumulative_returns)

    sim_df = pd.DataFrame(simulation_results).T

    # パーセンタイルで信頼区間を計算
    percentile_5  = sim_df.quantile(0.05, axis=1)
    percentile_50 = sim_df.quantile(0.50, axis=1)
    percentile_95 = sim_df.quantile(0.95, axis=1)

    print(f"最終資産（中央値）:   {percentile_50.iloc[-1]:.4f}倍")
    print(f"最終資産（下位5%）:   {percentile_5.iloc[-1]:.4f}倍")
    print(f"最終資産（上位95%）:  {percentile_95.iloc[-1]:.4f}倍")

    return pd.DataFrame({
        'p5':  percentile_5,
        'p50': percentile_50,
        'p95': percentile_95,
    })

# 使い方
# sim_results = monte_carlo_simulation(returns, n_simulations=1000, n_days=252)

Layer 1: データ取得層     ← 取引所APIからOHLCVデータを収集
Layer 2: 特徴量計算層     ← テクニカル指標・センチメント特徴量を計算
Layer 3: 予測モデル層     ← 機械学習モデル（LSTM / LightGBM）で価格方向を予測
Layer 4: シグナル生成層   ← 予測結果を売買シグナル（BUY/SELL/HOLD）に変換
Layer 5: リスク管理層     ← ポジションサイジング・ストップロス・最大損失制限
Layer 6: 注文執行層       ← 取引所APIへの実注文送信・約定確認

import numpy as np
from dataclasses import dataclass
from typing import Literal

@dataclass
class TradingSignal:
    direction:   Literal['BUY', 'SELL', 'HOLD']
    confidence:  float    # 0.0 ~ 1.0
    entry_price: float
    stop_loss:   float
    take_profit: float

def generate_signal(
    model_proba: np.ndarray,
    current_price: float,
    threshold_buy:   float = 0.60,
    threshold_sell:  float = 0.40,
    stop_loss_pct:   float = 0.02,
    take_profit_pct: float = 0.04,
) -> TradingSignal:
    """
    機械学習モデルの確率出力を売買シグナルに変換する

    Parameters:
    -----------
    model_proba     : [下落確率, 上昇確率] の配列
    current_price   : 現在価格
    threshold_buy   : BUYシグナルの閾値（デフォルト60%以上で買い）
    threshold_sell  : SELLシグナルの閾値（デフォルト40%以下で売り）
    stop_loss_pct   : ストップロスの割合（デフォルト2%）
    take_profit_pct : テイクプロフィットの割合（デフォルト4%）
    """
    up_proba = model_proba[1]  # 上昇確率

    if up_proba >= threshold_buy:
        direction   = 'BUY'
        stop_loss   = current_price * (1 - stop_loss_pct)
        take_profit = current_price * (1 + take_profit_pct)
    elif up_proba <= threshold_sell:
        direction   = 'SELL'
        stop_loss   = current_price * (1 + stop_loss_pct)
        take_profit = current_price * (1 - take_profit_pct)
    else:
        direction   = 'HOLD'
        stop_loss   = 0.0
        take_profit = 0.0

    return TradingSignal(
        direction=direction,
        confidence=abs(up_proba - 0.5) * 2,
        entry_price=current_price,
        stop_loss=stop_loss,
        take_profit=take_profit,
    )

import ccxt
from typing import Optional

class OrderExecutor:
    """
    ccxtを使った取引所への注文執行クラス
    """

    def __init__(self, exchange_id: str, api_key: str, secret: str):
        exchange_class = getattr(ccxt, exchange_id)
        self.exchange  = exchange_class({
            'apiKey':          api_key,
            'secret':          secret,
            'enableRateLimit': True,
        })

    def execute_signal(
        self,
        signal:   TradingSignal,
        symbol:   str,
        amount:   float,
        dry_run:  bool = True,  # デフォルトはペーパートレード
    ) -> Optional[dict]:
        """
        シグナルに基づいて注文を執行する

        Parameters:
        -----------
        signal  : TradingSignal（シグナル生成層からの出力）
        symbol  : 取引ペア（例："BTC/USDT"）
        amount  : 注文量（BTC単位）
        dry_run : Trueならペーパートレード（実際には注文しない）
        """
        if signal.direction == 'HOLD':
            return None

        side = 'buy' if signal.direction == 'BUY' else 'sell'

        if dry_run:
            print(f"[DRY RUN] {side.upper()} {amount} {symbol} @ {signal.entry_price}")
            print(f"          SL: {signal.stop_loss:.2f} | TP: {signal.take_profit:.2f}")
            return {'status': 'dry_run', 'side': side, 'amount': amount}

        # 実際の注文執行（本番環境のみ）
        order = self.exchange.create_order(
            symbol=symbol,
            type='market',
            side=side,
            amount=amount,
        )
        return order

def kelly_position_size(
    win_rate:       float,
    avg_win:        float,
    avg_loss:       float,
    capital:        float,
    kelly_fraction: float = 0.25,  # フルケリーの25%で保守的に
) -> float:
    """
    ケリー基準によるポジションサイズ計算

    Parameters:
    -----------
    win_rate       : 勝率（0.0 ~ 1.0）
    avg_win        : 平均利益率（例：0.04 = 4%）
    avg_loss       : 平均損失率（例：0.02 = 2%）
    capital        : 総資産額
    kelly_fraction : ケリー係数の適用割合（過剰投資防止）
    """
    if avg_loss == 0:
        return 0.0

    b = avg_win / avg_loss  # ペイオフレシオ
    p = win_rate            # 勝率
    q = 1 - win_rate        # 負け率

    kelly = (b * p - q) / b  # ケリー係数

    if kelly <= 0:
        return 0.0            # 期待値がマイナスなら投資しない

    # フルケリーは過剰リスクなので25%で適用
    position_pct  = kelly * kelly_fraction
    position_size = capital * position_pct

    return round(position_size, 2)

# 例：勝率55%、平均利益4%、平均損失2%、資産100万円の場合
# size = kelly_position_size(0.55, 0.04, 0.02, 1_000_000)

class RiskManager:
    """
    自動売買システムのリスク管理クラス
    機械学習モデルの出力に依存しない独立したリスク制御層
    """

    def __init__(
        self,
        max_drawdown_pct:   float = 0.15,   # 最大ドローダウン15%で全ポジション清算
        max_daily_loss_pct: float = 0.05,   # 日次最大損失5%でその日の取引停止
        max_position_pct:   float = 0.20,   # 1ポジション最大20%
    ):
        self.max_drawdown_pct   = max_drawdown_pct
        self.max_daily_loss_pct = max_daily_loss_pct
        self.max_position_pct   = max_position_pct
        self.peak_capital        = None
        self.daily_start_capital = None

    def check_trading_allowed(self, current_capital: float) -> tuple[bool, str]:
        """
        取引継続可否をチェックする

        Returns:
        --------
        (is_allowed, reason) : 取引可否とその理由
        """
        if self.peak_capital is None:
            self.peak_capital = current_capital

        if self.daily_start_capital is None:
            self.daily_start_capital = current_capital

        # ピーク比ドローダウン計算
        drawdown = (self.peak_capital - current_capital) / self.peak_capital
        if drawdown >= self.max_drawdown_pct:
            return False, f"最大ドローダウン超過: {drawdown:.2%}"

        # 日次損失チェック
        daily_loss = (self.daily_start_capital - current_capital) / self.daily_start_capital
        if daily_loss >= self.max_daily_loss_pct:
            return False, f"日次損失上限超過: {daily_loss:.2%}"

        # ピーク更新
        if current_capital > self.peak_capital:
            self.peak_capital = current_capital

        return True, "取引継続可"