データ取得 → シグナル生成 → リスク管理 → 注文執行 → 監視・通知

# Python 3.12以上を確認
python3 --version

# 仮想環境を作成・有効化
python3 -m venv venv
source venv/bin/activate  # Windowsは venv\Scripts\activate

# CCXTとCCXT Proをインストール
pip install ccxt "ccxt[pro]"

# 便利ライブラリも入れとく
pip install python-dotenv orjson pandas numpy

# .env ファイル（このファイル自体をGitに上げないこと！！ .gitignore に .env を追加すること）
EXCHANGE_API_KEY=your_api_key_here
EXCHANGE_API_SECRET=your_api_secret_here

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("EXCHANGE_API_KEY")
API_SECRET = os.getenv("EXCHANGE_API_SECRET")

if not API_KEY or not API_SECRET:
    raise ValueError(".env ファイルに EXCHANGE_API_KEY と EXCHANGE_API_SECRET を設定してください。")

# 絶対にコードに直書きしないこと！！
# API_KEY = "abc123..."  ← これは絶対ダメ！！

# public_api_example.py
import ccxt
import pandas as pd

# 取引所を初期化（Public APIはAPIキー不要）
exchange = ccxt.bitflyer()

# 1. Ticker（現在値）の取得
ticker = exchange.fetch_ticker('BTC/JPY')
print(f"現在値: {ticker['last']:,}円")
print(f"24h高値: {ticker['high']:,}円")
print(f"24h安値: {ticker['low']:,}円")

# 2. OHLCVローソク足データの取得（バックテストの基礎データ）
ohlcv = exchange.fetch_ohlcv('BTC/JPY', timeframe='1h', limit=100)
# [timestamp, open, high, low, close, volume] の形式で返ってくる
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
print(df.tail())

# 3. 板情報（オーダーブック）の取得
orderbook = exchange.fetch_order_book('BTC/JPY', limit=10)
print(f"最良買い気配: {orderbook['bids'][0]}")
print(f"最良売り気配: {orderbook['asks'][0]}")

# private_api_example.py
import ccxt
import os
from dotenv import load_dotenv

load_dotenv()

exchange = ccxt.bitflyer({
    'apiKey': os.getenv('EXCHANGE_API_KEY'),
    'secret': os.getenv('EXCHANGE_API_SECRET'),
})

# 残高確認
balance = exchange.fetch_balance()
jpy_balance = balance['JPY']['free']   # 利用可能なJPY
btc_balance = balance['BTC']['free']   # 利用可能なBTC
print(f"JPY残高: {jpy_balance:,.0f}円")
print(f"BTC残高: {btc_balance:.8f} BTC")

# 注文前の精度チェック（重要！！）
markets = exchange.load_markets()
market = markets['BTC/JPY']
print(f"最小注文量: {market['limits']['amount']['min']}")
print(f"価格精度: {market['precision']['price']}")

# order_execution.py
import ccxt
import os
from dotenv import load_dotenv

load_dotenv()

exchange = ccxt.bitflyer({
    'apiKey': os.getenv('EXCHANGE_API_KEY'),
    'secret': os.getenv('EXCHANGE_API_SECRET'),
})

def place_market_order(symbol: str, side: str, amount: float):
    """成行注文を発行する"""
    try:
        order = exchange.create_market_order(symbol, side, amount)
        print(f"成行{side}注文完了: {order['id']}")
        return order
    except ccxt.InsufficientFunds as e:
        print(f"残高不足: {e}")
        raise

def place_limit_order(symbol: str, side: str, amount: float, price: float):
    """指値注文を発行する"""
    try:
        order = exchange.create_limit_order(symbol, side, amount, price)
        print(f"指値{side}注文完了: {order['id']} @ {price:,}")
        return order
    except ccxt.InvalidOrder as e:
        print(f"注文パラメータエラー: {e}")
        raise

# 使用例
# place_market_order('BTC/JPY', 'buy', 0.001)          # 0.001 BTC を成行買い
# place_limit_order('BTC/JPY', 'buy', 0.001, 5000000)  # 500万円で指値買い

# error_handling.py
import ccxt
import time
import random
from functools import wraps

def exponential_backoff_retry(max_retries: int = 5, base_delay: float = 1.0):
    """指数バックオフによるリトライデコレーター"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except ccxt.RateLimitExceeded as e:
                    # レート制限：長めに待つ
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    print(f"レート制限に達しました。{delay:.1f}秒後にリトライ（{attempt+1}/{max_retries}）")
                    time.sleep(delay)
                except ccxt.NetworkError as e:
                    # ネットワークエラー：リトライ
                    delay = base_delay * (2 ** attempt)
                    print(f"ネットワークエラー: {e}。{delay:.1f}秒後にリトライ（{attempt+1}/{max_retries}）")
                    time.sleep(delay)
                except ccxt.ExchangeNotAvailable as e:
                    # 取引所メンテ・障害：長めに待つ
                    delay = base_delay * (2 ** attempt) * 2
                    print(f"取引所が利用不可: {e}。{delay:.1f}秒後にリトライ（{attempt+1}/{max_retries}）")
                    time.sleep(delay)
                except ccxt.AuthenticationError as e:
                    # 認証エラー：リトライ不要
                    print(f"認証エラー: {e}。APIキーを確認してください。")
                    raise
                except ccxt.BaseError as e:
                    # その他のCCXTエラー
                    print(f"CCXTエラー: {e}")
                    raise
            raise Exception(f"{max_retries}回リトライしましたが失敗しました。")
        return wrapper
    return decorator


class CircuitBreaker:
    """連続失敗時に一定時間APIコールを停止するサーキットブレーカー"""

    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.is_open = False

    def call(self, func, *args, **kwargs):
        if self.is_open:
            elapsed = time.time() - self.last_failure_time
            if elapsed < self.recovery_timeout:
                raise Exception(f"サーキットブレーカーOPEN中。あと{self.recovery_timeout - elapsed:.0f}秒待ってください。")
            else:
                self.is_open = False
                self.failure_count = 0

        try:
            result = func(*args, **kwargs)
            self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.is_open = True
                print(f"サーキットブレーカーが開きました。{self.recovery_timeout}秒間APIコールを停止します。")
            raise

# backtest_ma_cross.py
import ccxt
import pandas as pd
import numpy as np

def fetch_ohlcv_df(exchange_id: str, symbol: str, timeframe: str, limit: int = 500) -> pd.DataFrame:
    """OHLCVデータを取得してDataFrameに変換する"""
    exchange = getattr(ccxt, exchange_id)()
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df


def generate_ma_cross_signals(df: pd.DataFrame, short_window: int = 20, long_window: int = 50) -> pd.DataFrame:
    """移動平均クロスのシグナルを生成する"""
    df = df.copy()
    df['ma_short'] = df['close'].rolling(window=short_window).mean()
    df['ma_long']  = df['close'].rolling(window=long_window).mean()

    # シグナル生成：短期MAが長期MAを上回ったら買い(1)、下回ったら売り(-1)
    df['signal'] = 0
    df.loc[df['ma_short'] > df['ma_long'], 'signal'] = 1
    df.loc[df['ma_short'] <= df['ma_long'], 'signal'] = -1

    # ポジション変化点（クロス）を検出
    df['position'] = df['signal'].diff()
    return df


def calculate_pnl(df: pd.DataFrame, initial_capital: float = 1_000_000.0) -> pd.DataFrame:
    """損益を計算する"""
    df = df.copy()
    df['returns'] = df['close'].pct_change()
    df['strategy_returns'] = df['signal'].shift(1) * df['returns']

    # 取引コスト（片道0.1%）を差し引く
    trade_cost = 0.001
    df.loc[df['position'] != 0, 'strategy_returns'] -= trade_cost

    df['cumulative_returns'] = (1 + df['strategy_returns']).cumprod()
    df['portfolio_value'] = initial_capital * df['cumulative_returns']
    return df


# 実行例
if __name__ == '__main__':
    df = fetch_ohlcv_df('bitflyer', 'BTC/JPY', '1h', limit=500)
    df = generate_ma_cross_signals(df, short_window=20, long_window=50)
    df = calculate_pnl(df, initial_capital=1_000_000)
    print(df[['close', 'ma_short', 'ma_long', 'signal', 'portfolio_value']].tail(10))

# metrics.py
import numpy as np
import pandas as pd


def sharpe_ratio(strategy_returns: pd.Series, risk_free_rate: float = 0.0, periods_per_year: int = 8760) -> float:
    """シャープレシオを計算する（hourlyデータ想定でperiods_per_year=8760）"""
    excess_returns = strategy_returns - risk_free_rate / periods_per_year
    if excess_returns.std() == 0:
        return 0.0
    return float((excess_returns.mean() / excess_returns.std()) * np.sqrt(periods_per_year))


def max_drawdown(portfolio_values: pd.Series) -> float:
    """最大ドローダウンを計算する（0〜1の比率で返す）"""
    rolling_max = portfolio_values.cummax()
    drawdown = (portfolio_values - rolling_max) / rolling_max
    return float(drawdown.min())


def win_rate(strategy_returns: pd.Series) -> float:
    """勝率を計算する"""
    non_zero = strategy_returns[strategy_returns != 0]
    if len(non_zero) == 0:
        return 0.0
    return float((non_zero > 0).sum() / len(non_zero))


def print_metrics(df: pd.DataFrame) -> None:
    """評価指標をまとめて表示する"""
    sr  = sharpe_ratio(df['strategy_returns'].dropna())
    mdd = max_drawdown(df['portfolio_value'].dropna())
    wr  = win_rate(df['strategy_returns'].dropna())
    final_value  = df['portfolio_value'].iloc[-1]
    total_return = (final_value / df['portfolio_value'].iloc[0] - 1) * 100

    print("=" * 40)
    print("バックテスト評価指標")
    print("=" * 40)
    print(f"シャープレシオ    : {sr:.2f}")
    print(f"最大ドローダウン  : {mdd*100:.2f}%")
    print(f"勝率              : {wr*100:.1f}%")
    print(f"最終資産          : {final_value:,.0f}円")
    print(f"トータルリターン  : {total_return:.2f}%")
    print("=" * 40)

# /etc/systemd/system/trading-bot.service
[Unit]
Description=Crypto Trading Bot
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/trading-bot
ExecStart=/home/ubuntu/trading-bot/venv/bin/python main.py
Restart=always
RestartSec=10
EnvironmentFile=/home/ubuntu/trading-bot/.env
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable trading-bot
sudo systemctl start trading-bot

# 状態確認
sudo systemctl status trading-bot

# ログ確認
sudo journalctl -u trading-bot -f

# notifier.py
import os
import requests


def send_slack_notification(message: str, webhook_url: str = None) -> bool:
    """Slack Incoming Webhookで通知を送る"""
    url = webhook_url or os.getenv('SLACK_WEBHOOK_URL')
    if not url:
        print("SLACK_WEBHOOK_URL が設定されていません。")
        return False

    payload = {"text": message}
    try:
        response = requests.post(url, json=payload, timeout=10)
        response.raise_for_status()
        return True
    except requests.RequestException as e:
        print(f"Slack通知に失敗しました: {e}")
        return False


def notify_order_executed(side: str, amount: float, price: float) -> None:
    """注文実行を通知する"""
    emoji = "買" if side == "buy" else "売"
    message = f"[{emoji}い注文] {amount:.4f} BTC @ {price:,.0f}円"
    send_slack_notification(message)


def notify_error(error_message: str) -> None:
    """エラーを通知する"""
    send_slack_notification(f"[エラー] {error_message}")


def notify_daily_report(portfolio_value: float, daily_pnl: float) -> None:
    """日次レポートを通知する"""
    sign = "+" if daily_pnl >= 0 else ""
    message = (
        f"[日次レポート]\n"
        f"資産残高: {portfolio_value:,.0f}円\n"
        f"本日損益: {sign}{daily_pnl:,.0f}円"
    )
    send_slack_notification(message)