トレーディングボットプロジェクト
プロジェクト概要
本プロジェクトはトレーディングボットで、株式、外国為替、暗号通貨などの市場での取引戦略を自動化して実行します。
Pythonプログラミング言語を使用し、市場データの取得、戦略の実装、取引の実行、ログの記録、データベース管理など、多岐にわたる機能をカバーしています。
各コンポーネントは特定の機能を担い、自動取引の目標を達成するために協力しています。
ファイル構造
main.py: プロジェクトのエントリーポイント。トレーディングボットの初期化、設定の設定、メインループの開始。
settings.py: グローバル設定または設定パラメーターを含む。
database_manager.py: データベース関連の操作を管理。
logger.py: ログ記録機能を担当。
trade_executor.py: 実際の取引操作を実行。
evaluator.py: 市場条件または取引シグナルの評価ロジックを含む。
ohlcv_fetcher.py: OHLCVデータの取得。
multiple_timeframe_strategy.py: 複数のタイムフレームの取引戦略ロジック。
volatility.py: 市場の変動性を計算するロジック。
rsi.py: 相対力指数（RSI）の計算を実装。
これらのファイルは一緒に動作して、トレーディングボットの機能を実現します。各ファイルは独立した機能を持ち、main.pyを通じて統合されます。データベースの操作（database_manager.py）、ログ記録（logger.py）、市場データの取得（ohlcv_fetcher.py）、取引戦略の実行（multiple_timeframe_strategy.pyなど）がこのアーキテクチャの主要な部分です。
ワークフロー
起動プロセス（main.py）: `main.py`はプロジェクトの開始点です。
設定の初期化（`settings.py`によって提供される）、データベース接続の設定（`database_manager.py`を通じて）、ログの設定（`logger.py`を通じて）を担当します。
データ取得（ohlcv_fetcher.py）: 市場データ（例：OHLCV）を取得し、戦略分析に使用します。
市場分析（evaluator.py, volatility.py, rsi.py）: `evaluator.py`は取得した市場データを分析し、`volatility.py`で計算された市場の変動性や`rsi.py`で計算されたRSI指標を組み合わせて、現在の市場条件を評価する可能性があります。
戦略実行（multiple_timeframe_strategy.py）: 市場分析の結果に基づき、`multiple_timeframe_strategy.py`で定義された取引戦略が取引を実行するかどうかを決定します。
取引実行（trade_executor.py）: 取引シグナルが確認されると、`trade_executor.py`が市場に取引注文を送信します。
ログとデータ保存（logger.py, database_manager.py）: 取引活動とシステム状態はログに記録され、重要なデータはデータベースに保存され、後続の分析に利用されます。
使用方法
Python環境がインストールされていることを確認してください。
必要な依存関係をインストールします。
`settings.py`のパラメーターを設定します（APIキー、データベース設定など）。
`main.py`を実行してトレーディングボットを起動します。
注意点
このトレーディングボットは教育および研究目的でのみ使用してください。
実際の取引でこのボットを使用する前に、十分にテストし、そのリスクを理解してください。
プロジェクトのPythonファイルに含まれる依存関係：
`ccxt` - 暗号通貨取引のためのライブラリ。
`config` - 設定管理用。
`data_fetcher` - データ取得のためのカスタムモジュール。
`database_manager` - データベース管理のためのカスタムモジュール。
`date_evaluator` - 日付と時刻の評価のためのカスタムモジュール。
`datetime` - 日付と時刻の操作のための標準Pythonモジュール。
`indicators` - 技術指標のためのカスタムモジュール。
`numpy` - 数値計算のためのライブラリ。
`pandas` - データ処理と分析のためのライブラリ。
`position` - 取引位置の管理のためのカスタムモジュール。
`random` - ランダム数を生成するための標準Pythonモジュール。
`sqlite3` - SQLiteデータベース操作のためのライブラリ。
`strategies` - 取引戦略の定義のためのカスタムモジュール。
`tabulate` - テーブルデータを印刷するためのライブラリ。
`time` - 時間操作のための標準Pythonモジュール。
`trading` - 取引ロジックのためのカスタムモジュール。
`utils` - 様々なユーティリティのためのカスタムモジュール。
`numpy`、`pandas`、`ccxt`、`datetime`などは一般的なサードパーティライブラリで、pipなどのパッケージマネージャーを通じてインストールできます。`config`、`data_fetcher`、`database_manager`などの他のものは、プロジェクト固有のカスタムモジュールである可能性があります。依存関係をインストールする際には、これらのカスタムモジュールが通常のパッケージマネージャーを通じてインストールできない可能性があることに特に注意が必要です。
`main.py` ファイルのコード分析
インポートするモジュール
標準ライブラリ: `time`, `datetime`
サードパーティライブラリ: `ccxt`, `numpy`, `tabulate`
カスタムモジュール:
`config`
`strategies.multiple_timeframe_strategy`
`trading.trade_executor`
`trading.position`
`utils.logger`
`date_evaluator.evaluator`
`database_manager`
クラス定義
MockAccount:
模擬取引アカウントのクラス。
購入と販売のトレードを実行し、残高とポジションを管理します。
関数定義
check_profit_and_exit:
ポジションが利益目標に達したかどうかをチェックし、達していれば終了します。
fetch_highs_lows:
異なるタイムフレームの高値と安値を取得します。
main:
メイン関数で、プログラムの主要なロジックを実行します。
主なロジック
設定の初期化と取引所への接続、模擬アカウントの設定。
ループ処理を行い、各ループで以下のステップを実行：
市場データを取得。
複数のタイムフレーム戦略を使用して市場を分析。
トレード（購入・販売）を実行。
トレードログを記録。
終了条件が満たされたかどうかをチェック。
一定の時間間隔を待って次のループに進む。
特徴と機能
`ccxt`ライブラリを使用して暗号通貨取引所に接続。
複数のタイムフレーム戦略を利用して市場を分析し、トレード決定を行う。
模擬アカウントを使用してトレード行動をシミュレート。
トレード実行とログ記録機能。
教育および研究目的のためのトレーディングボット。
このファイルはトレーディングボットシステムの中核で、異なるモジュールの調整、取引戦略の実行、トレード活動の記録、市場データの定期的な更新と分析を担当します。
main.pyソースコード
import time # 標準ライブラリtimeをインポート
from datetime import datetime # datetimeライブラリからdatetimeをインポート
import ccxt # ccxtライブラリをインポート（暗号通貨取引所への接続に使用）
import numpy as np # numpyライブラリをインポート（数値計算に使用）
from tabulate import tabulate # tabulateライブラリをインポート（テーブル形式のデータ表示に使用）
# カスタムモジュールのインポート
from config import settings # 設定情報を含むconfigモジュールをインポート
from strategies.multiple_timeframe_strategy import get_multiple_timeframe_indicators # 複数の時間枠に基づく指標を取得する関数をインポート
from trading.trade_executor import execute_trade # トレード実行機能を提供するモジュールをインポート
from trading.position import Position # トレードのポジションを管理するクラスをインポート
from utils.logger import save_log_to_csv # ログ記録機能を提供するモジュールをインポート
import date_evaluator.evaluator as evaluator # 日付評価機能を提供するモジュールをインポート
# データベース管理モジュールのインポート
import database_manager # データベース操作に関連する機能を提供するモジュールをインポート
# 模擬取引アカウントのクラス
class MockAccount:
def __init__(self, initial_balance): # 初期化関数
self.balance = initial_balance # 残高の初期化
self.positions = [] # ポジションのリスト
self.trade_history = [] # トレード履歴のリスト
def execute_trade(self, symbol, trade_type, amount, price): # トレード実行関数
if trade_type == "buy":
cost = amount * price
if cost <= self.balance:
self.balance -= cost # 残高からコストを減算
self.positions.append((symbol, amount, price)) # ポジションに追加
self.trade_history.append(('buy', symbol, amount, price)) # トレード履歴に追加
return "Buy executed" # 購入実行のメッセージ
else:
return "Insufficient balance" # 残高不足のメッセージ
elif trade_type == "sell":
for position in self.positions:
if position[0] == symbol and position[1] >= amount:
self.positions.remove(position) # ポジションから削除
self.balance += amount * price # 残高に加算
self.trade_history.append(('sell', symbol, amount, price)) # トレード履歴に追加
return "Sell executed" # 販売実行のメッセージ
return "Position not found" # ポジションが見つからない場合のメッセージ
else:
return "Invalid trade type" # 無効なトレードタイプの場合のメッセージ
# 利益と損失の確認関数
def check_profit_and_exit(position, profit_target): # ポジションと利益目標を引数にとる
if position.profit >= profit_target:
print(f"利益目標達成：{position.profit} >= {profit_target}")
return True # 目標達成の場合はTrueを返す
print(f"利益目標未達：{position.profit} < {profit_target}")
return False # 目標未達の場合はFalseを返す
# 高値と安値を取得する関数
def fetch_highs_lows(exchange_id=settings.EXCHANGE_ID, symbol=settings.SYMBOL, timeframes=settings.TIMEFRAMES, limit=settings.LIMIT):
exchange = getattr(ccxt, exchange_id)() # 指定された取引所のオブジェクトを生成
highs_lows = {} # 高値と安値を格納する辞書
for tf in timeframes:
ohlcv = exchange.fetch_ohlcv(symbol, tf, limit=limit) # OHLCVデータの取得
highs = [x[2] for x in ohlcv] # 高値リスト
lows = [x[3] for x in ohlcv] # 安値リスト
highs_lows[tf] = {
'high': max(highs), # 最高値
'low': min(lows) # 最低値
}
return highs_lows # 高値と安値の辞書を返す
# フィボナッチリトレースメントの計算関数
def calculate_fibonacci_retracement(high, low, precision=4):
difference = high - low # 高値と安値の差
fib_levels = {} # フィボナッチレベルを格納する辞書
for level in settings.FIBONACCI_LEVELS:
retracement = high - difference * level # リトレースメントの計算
retracement_value = round(retracement, precision) # 小数点以下の丸め
level_label = f"{round(level * 100)}%" # レベルのラベル
fib_levels[level_label] = retracement_value # 辞書に追加
fib_levels['100%'] = low # 100%レベルの追加
return fib_levels # フィボナッチレベルの辞書を返す
# RSI指標の評価関数
def rsi_rating(rsi): # RSI値を引数にとる
if rsi > 70:
return "超買い" # RSIが70を超える場合は「超買い」と評価
elif rsi < 30:
return "超売り" # RSIが30未満の場合は「超売り」と評価
else:
return "中立" # それ以外の場合は「中立」と評価
# 各時間帯に対する取引の評価
# 各時間帯に応じて取引に適した時期かどうかを判断する
def evaluate_time_period(hour):
if 23 <= hour or hour < 1:
return "夜間：取引に不向き"
elif 1 <= hour < 3:
return "深夜：取引に不向き"
elif 3 <= hour < 5:
return "早朝：取引に不向き"
elif 7 <= hour < 9:
return "朝：取引に適している"
# 以下、他の時間帯の評価が続く...
# 最新の価格を取得する関数
def fetch_latest_price(exchange_id=settings.EXCHANGE_ID, symbol=settings.SYMBOL):
exchange = getattr(ccxt, exchange_id)()
ticker = exchange.fetch_ticker(symbol)
return ticker['last']
# 取引実行の判断を行う関数
# 指標、日付タイプ、時間帯のアドバイスに基づいて取引の実行可否を判断
def should_execute_trade(indicators, day_type, time_advice):
rsi_series = indicators['2h']['rsi_history']
if not rsi_series.empty:
rsi_value = rsi_series.iloc[-1]
is_oversold = rsi_value < settings.RSI_THRESHOLD
is_trading_day = "休日" not in day_type
is_trading_time = "取引に適している" in time_advice
if is_oversold and is_trading_day and is_trading_time:
return True, "取引条件に適合、売り提案"
else:
return False, "取引条件に不適合、市場分析を推奨"
return False, "データ不足で取引判断不可"
# メイン関数
def main():
mock_account = MockAccount(10000) # 初期資金 10000
capital = settings.CAPITAL
profit_target = settings.PROFIT_TARGET
log_data = []
while True:
today = datetime.now().strftime("%Y-%m-%d")
current_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 日付評価モジュールによる今日の評価
day_info = evaluator.evaluate_date(today)
general_advice = evaluator.get_advice(day_info[0])
# 現在の時間帯の評価
time_advice = evaluate_time_period(datetime.now().hour)
print(time_advice)
# 最新のビットコイン価格の取得
latest_price = fetch_latest_price()
print(f"ビットコイン最新価格: {latest_price}")
# 複数時間枠の指標取得
indicators = get_multiple_timeframe_indicators(settings.SYMBOL, settings.TIMEFRAMES, settings.LIMIT)
# 取引実行の判断とアドバイス
trade_decision, trade_advice = should_execute_trade(indicators, day_info[0], time_advice)
print(trade_advice)
if trade_decision:
# ここに取引実行のロジックを記述
pass
# 各時間枠の高値と安値の取得
highs_lows = fetch_highs_lows()
for tf, values in highs_lows.items():
print(f"{tf} 時間枠 - 最高値: {values['high']}, 最低値: {values['low']}")
for tf in settings.TIMEFRAMES:
if tf in highs_lows:
high = highs_lows[tf]['high']
low = highs_lows[tf]['low']
fib_levels = calculate_fibonacci_retracement(high, low)
print(f"{tf} 時間枠 - フィボナッチリトレースメントレベル: {fib_levels}")
indicators = get_multiple_timeframe_indicators(settings.SYMBOL, settings.TIMEFRAMES, settings.LIMIT)
current_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# ログエントリの設定
log_entry = {
"btc_latest_price": latest_price, # 最新のビットコイン価格
"day_info": day_info[0], # 日付情報
"advice": general_advice, # アドバイス
}
# 各時間枠の高値と安値をログエントリに追加
for tf in settings.TIMEFRAMES:
if tf in highs_lows:
log_entry[f"high_{tf}"] = highs_lows[tf]['high']
log_entry[f"low_{tf}"] = highs_lows[tf]['low']
# RSI値とボラティリティ値のリストの初期化
rsi_values = []
volatility_values = []
# 各時間枠の指標データをログエントリに追加
for tf, values in indicators.items():
log_entry[f"rsi_{tf}_history"] = [round(val, 3) for val in list(values['rsi_history'])]
log_entry[f"volatility_{tf}_history"] = [round(val, 3) for val in list(values['volatility_history'])]
# 各指標の最高値と最低値をログエントリに追加
log_entry[f"rsi_{tf}_high"] = round(values['rsi_high'], 3)
log_entry[f"rsi_{tf}_low"] = round(values['rsi_low'], 3)
log_entry[f"volatility_{tf}_high"] = round(values['volatility_high'], 3)
log_entry[f"volatility_{tf}_low"] = round(values['volatility_low'], 3)
# RSIシリーズから最新のRSI値を取得し、リストに追加
rsi_series = values['rsi_history']
if not rsi_series.empty:
rsi_value = rsi_series.iloc[-1]
rsi_values.append(round(rsi_value, 3))
rating = rsi_rating(rsi_value)
log_entry[f"rsi_{tf}_rating"] = rating
# RSIとボラティリティの平均値を計算し、ログエントリに追加
log_entry['average_rsi'] = round(np.mean(rsi_values), 3) if rsi_values else np.nan
log_entry['average_volatility'] = round(np.mean(volatility_values), 3) if volatility_values else np.nan
log_entry['trade_result'] = 0
# ログデータにログエントリを追加し、データベースに記録
log_data.append(log_entry)
database_manager.log_to_db(log_data)
# 特定の期間におけるRSIの評価統計を取得
start_date = "2023-11-25"
end_date = "2023-11-25"
rsi_rating_stats = database_manager.query_rsi_ratings(start_date, end_date)
# RSI評価統計の表示
if rsi_rating_stats:
# RSI評価統計の詳細を表示するコードが続く...
pass
# ログエントリから不要なキーを削除
keys_to_remove = [
# 削除するキーのリスト
]
for key in keys_to_remove:
log_entry.pop(key, None)
# ログエントリのキーをリスト化して、それぞれのサブログを作成し、表示する
keys = list(log_entry.keys())
for i in range(0, len(keys), 4):
sub_log = {key: log_entry[key] for key in keys[i:i+4]}
# tabulateライブラリを使用して、サブログを表形式で表示
print(tabulate([sub_log], headers="keys", tablefmt="grid"))
# ログデータをリセット
log_data = []
# 設定された時間間隔でスリープ（待機）
time.sleep(settings.SLEEP_INTERVAL)
# メイン関数の実行
if __name__ == "__main__":
main()
settings.py 作業中
CAPITAL: 10000 # 初期資金
PROFIT_TARGET: 100 # 利益目標（例値）
SYMBOL: 'BTC/USDT' # 取引ペア
TIMEFRAMES: ['5m', '15m', '1h', '2h'] # 時間枠
LIMIT: 18 # 過去18本のK線データを取得するための上限
EXCHANGE_ID: 'binance' # 取引所ID
LOG_FILE_NAME: "trading_log.csv" # ログファイル名
RSI_THRESHOLD: 30 # RSIの閾値（決定のために使用）
VOLATILITY_THRESHOLD: 0.05 # ボラティリティの閾値（決定のために使用）
SLEEP_INTERVAL: 60 # メインループの待機時間（秒）
FIBONACCI_LEVELS: [0.236, 0.382, 0.5, 0.618] # 使用するフィボナッチリトレースメントレベル
INITIAL_MOCK_BALANCE: 10000 # 模擬アカウントの初期資金
TRADE_AMOUNT: 1 # 各取引での固定ビットコイン数量（または他の資産）
MAX_POSITION_TIME: 3600 # 最大保有時間（秒）
MAX_POSITIONS: 5 # 最大保有ポジション数
この記事が気に入ったらサポートをしてみませんか？