- 免責事項: このコードは教育・テスト目的のみで提供します。実際の取引では資金を失うリスクがあります。バックテストを行い、BybitのAPI制限や手数料を考慮してください。私は投資アドバイスを提供していません。APIキーなどの機密情報は絶対に共有せず、安全に扱ってください。
- 前提: Python環境にccxt、pandas、numpy、ta-lib(ボリンジャーバンド計算用)をインストールする必要があります(pip install ccxt pandas numpy TA-Lib)。TA-Libはオプションで、pandas/numpyだけで計算可能。
- 戦略の概要: シンプルなボリンジャーバンド戦略を採用。
- ボリンジャーバンド: 期間20、標準偏差2。
- 買いシグナル: 価格が下バンドを下回ったらロングポジション(買い)。
- 売りシグナル: 価格が上バンドを上回ったらショートポジション(売り)、またはポジションをクローズ。
- 4時間足データを定期的に取得し、チェック。
- Bybit API: ccxtライブラリを使ってBybitを操作。BybitアカウントでAPIキーを作成し、コードに設定。
- 実行方法: スクリプトをループで実行(例: 無限ループにsleepを入れて4時間ごとにチェック)。テスト時はデモモード(Bybitのテストネット)をおすすめ。
以下はサンプルコードです。コメントで説明を追加しています。
import ccxt
import pandas as pd
import numpy as np
import time
from datetime import datetime
# Bybit APIの設定(あなたのAPIキーとシークレットをここに入力)
exchange = ccxt.bybit({
'apiKey': 'YOUR_API_KEY',
'secret': 'YOUR_API_SECRET',
'enableRateLimit': True,
'options': {
'defaultType': 'swap', # 永久先物(USDTペア用)
}
})
# シンボルと時間足の設定
symbol = 'BTC/USDT:USDT' # BTC/USDT永久先物
timeframe = '4h' # 4時間足
bollinger_period = 20 # ボリンジャーバンドの期間
bollinger_dev = 2.0 # 標準偏差の倍率
position_size = 0.001 # 注文サイズ(BTC単位、調整してください)
# ボリンジャーバンドを計算する関数
def calculate_bollinger_bands(df, period=20, dev=2.0):
df['MA'] = df['close'].rolling(window=period).mean()
df['std'] = df['close'].rolling(window=period).std()
df['upper'] = df['MA'] + (df['std'] * dev)
df['lower'] = df['MA'] - (df['std'] * dev)
return df
# メインのトレーディングループ
while True:
try:
# OHLCVデータを取得(最新の100本の4時間足)
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=100)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
# ボリンジャーバンド計算
df = calculate_bollinger_bands(df, bollinger_period, bollinger_dev)
# 最新の価格とバンド
current_price = df['close'].iloc[-1]
upper_band = df['upper'].iloc[-1]
lower_band = df['lower'].iloc[-1]
ma = df['MA'].iloc[-1]
print(f"{datetime.now()}: Current Price: {current_price}, Upper: {upper_band}, Lower: {lower_band}")
# 現在のポジションを確認
positions = exchange.fetch_positions([symbol])
current_position = next((p for p in positions if p['symbol'] == symbol), None)
position_amount = current_position['contracts'] if current_position else 0 # ポジションサイズ
# 買いシグナル: 価格が下バンドを下回ったらロング(買い)
if current_price < lower_band and position_amount <= 0:
print("Buy Signal: Opening Long Position")
order = exchange.create_market_buy_order(symbol, position_size)
print(f"Order: {order}")
# 売りシグナル: 価格が上バンドを上回ったらショート(売り)またはクローズ
elif current_price > upper_band and position_amount >= 0:
print("Sell Signal: Opening Short Position")
order = exchange.create_market_sell_order(symbol, position_size)
print(f"Order: {order}")
# 他の条件(例: MAクロスでクローズなど)を追加可能
except Exception as e:
print(f"Error: {e}")
# 4時間ごとにチェック(sleep 4時間 = 14400秒、テスト時は短く調整)
time.sleep(14400)
説明とカスタマイズ
- データ取得: fetch_ohlcvで4時間足のOHLCV(Open, High, Low, Close, Volume)を取得。
- ボリンジャーバンド: pandasで移動平均(MA)と標準偏差を計算。
- 注文実行: create_market_buy_order / create_market_sell_orderでマーケット注文。レバレッジやストップロスを追加可能。
- 改善点:
- バックテスト: 過去データを取得してシミュレーション。
- リスク管理: ストップロス/テイクプロフィット追加。
- エラー処理: ネットワークエラー対応。
- TA-Lib使用: import talibしてupper, ma, lower = talib.BBANDS(df[‘close’], timeperiod=20, nbdevup=2, nbdevdn=2)で簡単に計算。
- テスト: Bybitのテストネット(testnet.bybit.com)でAPIを作成し、exchange.urls[‘api’] = exchange.urls[‘test’]を追加してテスト。
このコードを基に、具体的な戦略の変更(例: バンドタッチ後の反転確認)が必要なら教えてください!