PythonによるCFTCデータマイニングとAIモデルの構築

FX市場で成功するためには、テクニカル分析だけでなくファンダメンタル要因の考慮も必要です。あまり注目されていませんが価値の高い情報源としてCFTCレポート（COTおよびTFF）があり、これらは主要な市場参加者のポジションを明らかにし、機関投資家の行動を評価することを可能にします。

外国為替市場は世界最大の市場ですが、その高いボラティリティのため予測は困難です。COT/TFFレポートはスマートマネーの動向を把握する手がかりを提供し、隠れた市場トレンドの発見に役立ちます。

提案するアプローチは、COT/TFFデータと市場価格をPythonモデルに統合し、MetaTrader 5による自動取引と組み合わせるものです。これにより、分析から実行へ、遅延や人手の介入なしに移行できます。

理論的基礎

COTレポートとTFFレポートとは何か

外国為替市場における最大級のプレーヤー、すなわち数十億ドル規模の資産を運用するヘッジファンド、年金基金、投資銀行のポートフォリオの中身を知ることができると想像してください。これこそが、米国商品先物取引委員会(CFTC)が毎週金曜日に公表するCOT/TFFレポートによって得られる情報なのです。

これらのレポートは、1970年代から1980年代の市場危機の後に生まれました。当時、規制当局は市場参加者が大口プレーヤーの動向を把握するための情報を必要としていることに気づきました。現在では、一定の基準以上のポジションを保有するすべての参加者は、そのポジションを開示することが義務付けられています。CFTCはこの情報を集計し、毎週火曜日時点のポジション状況として公表しています。

市場の構成要素：参加者とその動機

商業トレーダーは、投機目的ではなくリスクヘッジのために先物を利用する実需の企業です。航空会社は燃料価格を固定するために原油先物を購入し、輸出企業は為替レート下落に備えて通貨先物を売却し、農家は収穫価格を事前に確保するために小麦先物を売却します。

非商業トレーダーは価格変動から利益を得るために取引する投機家であり、ヘッジファンドや投資銀行、大規模運用会社などが含まれます。彼らは市場を深く分析する能力を持つため「スマートマネー」と呼ばれることが多いです。

小口トレーダーとは、報告基準未満のポジションしか持たないその他すべての参加者を指します。個人投資家や小規模ファンド、個人投機家などが含まれます。

COTレポートが示すもの

COTレポートはすべての先物市場の全体像を提供します。各参加者グループごとに、ロングポジション（上昇への賭け）、ショートポジション（下落への賭け）、およびネットポジション（ロングとショートの差）が示されます。オープンインタレストは市場に存在する未決済契約の総数を示します。

たとえば、非商業トレーダーがユーロに対して+120,000契約のネットロングを持っている場合、投機家全体としてEURUSDの上昇を見込んでいることを意味します。一方、商業トレーダーが-115,000契約を示している場合、ヘッジャーはユーロの下落を予想しているか、あるいは単に通貨リスクを回避していることを意味します。

TFFレポートが追加するもの

TFFレポートは金融先物のみに焦点を当て、より詳細な機関投資家のポジションを提供します。ここでは非商業トレーダーがさらに細分化され、レバレッジファンド（借入を積極的に利用する攻撃的なヘッジファンド）とアセットマネージャー（年金基金や保険会社などの保守的な機関投資家）に分類されます。また、ディーラーや仲介業者といった市場流動性を提供する銀行も含まれます。

この詳細さは極めて重要です。もしレバレッジファンドがUSDのロングポジションを急増させた場合、それは短期的な投機需要のシグナルです。一方でアセットマネージャーが同様の動きを見せる場合、それは長期的な機関投資家のセンチメント変化を示しています。

トレーダーにとって必要な理由

レポートの主な価値は、価格に反映される前の「スマートマネー」の動きを示す点にあります。大口投資家は一般投資家がアクセスできない情報に基づいて行動することが多いです。

極端なポジションの検出では、投機家が記録的なロングポジションを取ると市場は買われすぎとなり、調整が起こりやすくなります。逆の場合は反発の可能性を示します。これは逆張りの原則に基づきます。

トレンド確認では、価格が上昇し非商業トレーダーがロングを増やし続ける場合、トレンドは強いと判断されます。一方で価格上昇時にポジションを減らし始める場合、トレンドは弱まっています。

早期反転検出では、価格とポジションの乖離が反転の前兆となることがあります。価格は上昇を続ける可能性がありますが、主要市場参加者がすでに強気エクスポージャーを減らしている場合、反転の兆候と考えられます。

実例

たとえばEURUSDのレポートで、レバレッジファンドが+85,000契約、アセットマネージャーが+25,000契約、ディーラーが-95,000契約を保有している場合、これは、積極的な投機筋と保守的な機関投資家の双方がユーロ上昇に賭けていることを意味し、総合シグナルとしてはユーロに強気です。ディーラーはその反対側のポジションを取り、おそらく顧客注文の処理や自らのリスクヘッジをおこなっています。総合的にはユーロに強気なシグナルと解釈できます。

重要な注意点

このデータは3日遅れで公表されるため、変動の激しい市場ではこの遅延が大きな意味を持ちます。すべての商業ポジションが純粋なヘッジとは限らず、一部の銀行は商業活動を装って投機をおこなう場合もあります。アルゴリズム取引と機械学習は参加者の従来の行動パターンを変化させています。

現代のアルゴリズムはCOT/TFFレポート自体を意思決定に利用することができ、これにより自己実現的な予言が生じ、データ解釈はさらに複雑になります。

これが有効な理由

COTおよびTFFレポートが機能する理由は、意見や予測ではなく実際の資金ポジションを示しているためです。ヘッジファンドが特定の市場方向に数十億ドルを投じる場合、それはあらゆる公開コメントよりもはるかに重要なシグナルとなります。このデータは、最も情報と資金力を持つ市場参加者の集合的な知見を反映しています。

機械学習および自動売買と組み合わせることで、機関投資家のポジション分析は、実際に市場を動かしている大口参加者の行動に基づく取引戦略を構築するための強力なツールとなります。

COT/TFFデータを用いた価格予測

価格予測は、機関投資家のポジションが将来の価格変動と相関するという仮定に基づいています。たとえば、投機筋のロングポジション増加は強気市場のシグナルとなり、ショートポジション増加は弱気センチメントを示す可能性があります。このデータは機械学習モデルの特徴量として使用され、MetaTrader 5から取得した過去価格データや、ボラティリティ・移動平均などのテクニカル指標と組み合わされます。このアプローチにより、市場センチメントと価格トレンドの間にある複雑な関係を捉えることが可能になります。

データ準備

COTおよびTFFレポートは、CFTCのウェブサイト上でExcel形式で公開されています。これらのデータの読み込みにはrequestsライブラリが使用され、処理にはpandasが使用されます。以下のコードは、パフォーマンスを改善するためにキャッシュを用いてCOTデータを読み込み、前処理する方法を示しています。

import requests
import zipfile
import pandas as pd
import os
import logging
import MetaTrader5 as mt5
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
import importlib.util
import glob
import asyncio

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

plt.style.use('ggplot')
plt.rcParams['figure.figsize'] = [10, 6]

COT_URL = "https://www.cftc.gov/files/dea/history/dea_fut_xls_2024.zip"
TFF_URL = "https://www.cftc.gov/files/dea/history/fut_fin_xls_2024.zip"
OUTPUT_DIR = "data"
os.makedirs(OUTPUT_DIR, exist_ok=True)

def load_cot_reports() -> pd.DataFrame:
    cache_path = os.path.join(OUTPUT_DIR, "cot_report.csv")
    if os.path.exists(cache_path):
        logger.info(f"Loading COT data from cache: {cache_path}")
        return pd.read_csv(cache_path)

    try:
        response = requests.get(COT_URL)
        response.raise_for_status()
        zip_path = os.path.join(OUTPUT_DIR, "cot_data.zip")
        with open(zip_path, "wb") as f:
            f.write(response.content)

        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            zip_ref.extractall(OUTPUT_DIR)
            excel_file = [f for f in zip_ref.namelist() if f.endswith('.xls') or f.endswith('.xlsx')][0]
        
        cot_data = pd.read_excel(os.path.join(OUTPUT_DIR, excel_file))
        relevant_columns = [
            "Market_and_Exchange_Names",
            "NonComm_Positions_Long_All",
            "NonComm_Positions_Short_All",
            "Comm_Positions_Long_All",
            "Comm_Positions_Short_All",
            "Open_Interest_All"
        ]
        cot_data = cot_data[relevant_columns]
        cot_data["Net_NonComm"] = cot_data["NonComm_Positions_Long_All"] - cot_data["NonComm_Positions_Short_All"]
        cot_data["Net_Comm"] = cot_data["Comm_Positions_Long_All"] - cot_data["Comm_Positions_Short_All"]
        cot_data.to_csv(cache_path, index=False)
        logger.info(f"COT report saved in {cache_path}")
        return cot_data
    except Exception as e:
        logger.error(f"Error loading COT data: {e}")
        return pd.DataFrame()

同様のプロセスはTFFレポートにも適用され、通貨先物でフィルタリングし、レバレッジファンドおよびアセットマネージャーのネットポジションを計算します。データキャッシュによりネットワークリクエストが最小化され、再処理の速度が向上します。

MetaTrader 5との統合

過去価格はMetaTrader 5ライブラリを通じてダウンロードされ、マーケットデータへのアクセスが提供されます。以下のコードは、過去30日間の1時間足データを取得する方法を示しています。

def get_historical_prices(pair: str, days_history: int = 30) -> pd.DataFrame:
    if not mt5.initialize():
        logger.error("MT5 initialization failed")
        return pd.DataFrame()

    timeframe = mt5.TIMEFRAME_H1
    utc_from = datetime.now() - timedelta(days=days_history)
    rates = mt5.copy_rates_from(pair, timeframe, utc_from, 24 * days_history)
    
    if rates is None or len(rates) == 0:
        logger.warning(f"No historical data for {pair}")
        return pd.DataFrame()

    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s')
    df.set_index('time', inplace=True)
    df['price_change_24h'] = df['close'].shift(-24) / df['close'] - 1
    df.dropna(inplace=True)
    return df[['open', 'high', 'low', 'close', 'tick_volume', 'price_change_24h']]

データ処理と統合

COT、TFF、および過去の価格データを統合し、派生特徴量を追加します。具体的には、ボラティリティ（高値と安値の差を終値で正規化したもの）、価格および出来高の24時間移動平均、さらにCOTおよびTFFポジションのラグおよび変化率を含みます。

以下は特徴量準備コードです。

def prepare_features(pair: str, cot_data: pd.DataFrame, tff_data: pd.DataFrame) -> pd.DataFrame:
    df_prices = get_historical_prices(pair)
    if df_prices.empty:
        logger.warning(f"No price data available for {pair}")
        df = pd.DataFrame(index=[datetime.now()], columns=['close', 'price_change_24h'])
        df['close'] = 1.0
        df['price_change_24h'] = 0.0
    else:
        df = df_prices.copy()

    df['volatility'] = (df['high'] - df['low']) / df['close']
    df['volume_sma_24'] = df['tick_volume'].rolling(window=24).mean()
    df['price_sma_24'] = df['close'].rolling(window=24).mean()
    df['price_change_1h'] = df['close'].pct_change()

    market = map_pair_to_cot_tff(pair)
    if market and not cot_data.empty:
        cot_subset = cot_data[cot_data["Market_and_Exchange_Names"].str.contains(market, case=False, na=False)]
        if not cot_subset.empty:
            cot_features = cot_subset[['Net_NonComm', 'Net_Comm']].mean().to_frame().T
            for col in cot_features.columns:
                df[col] = cot_features[col].iloc[0]

    if market and not tff_data.empty:
        tff_subset = tff_data[tff_data["Market_and_Exchange_Names"].str.contains(market, case=False, na=False)]
        if not tff_subset.empty:
            tff_features = tff_subset[['Net_Lev_Money', 'Net_Asset_Mgr']].mean().to_frame().T
            for col in tff_features.columns:
                df[col] = tff_features[col].iloc[0]

    for col in ['Net_NonComm', 'Net_Comm', 'Net_Lev_Money', 'Net_Asset_Mgr']:
        if col in df.columns:
            df[f'{col}_lag1'] = df[col].shift(1)
            df[f'{col}_change'] = df[col].pct_change().fillna(0)

    df.dropna(inplace=True)
    return df

def map_pair_to_cot_tff(pair: str) -> str:
    mapping = {
        'EURUSD': 'EURO FX',
        'GBPUSD': 'BRITISH POUND',
        'USDJPY': 'JAPANESE YEN',
        'AUDUSD': 'AUSTRALIAN DOLLAR',
        'USDCAD': 'CANADIAN DOLLAR',
        'USDCHF': 'SWISS FRANC',
        'NZDUSD': 'NEW ZEALAND DOLLAR'
    }
    base_pair = pair.replace('.ecn', '')[:6]
    return mapping.get(base_pair, '')

精度を向上させるために、木の数や最大深さといったハイパーパラメータを最適化することができ、またクロスバリデーションを用いてモデルのロバスト性を評価することができます。特徴量の重要度はCSVファイルに保存され、予測への影響を分析するために使用されます。

予測と可視化

価格予測

予測は最新データに基づいておこなわれ、24時間後の価格変化率を予測します。この予測は現在の市場価格に適用されます。以下が予測コードです。

async def get_price_forecast(pair: str, model, scaler) -> dict:
    df = prepare_features(pair, cot_data, tff_data)
    if df.empty:
        logger.warning(f"No data to forecast for {pair}")
        return {'pair': pair, 'forecast_price': None, 'confidence': 0.0}

    X_latest = df.drop(columns=['price_change_24h']).iloc[-1:]
    X_scaled = scaler.transform(X_latest)
    
    price_change_pred = model.predict(X_scaled)[0]
    confidence = model.score(X_scaled, df['price_change_24h'].iloc[-1:]) if len(df) > 1 else 0.6
    
    tick = mt5.symbol_info_tick(pair)
    if not tick:
        logger.warning(f"No current data for {pair}")
        return {'pair': pair, 'forecast_price': None, 'confidence': 0.0}
    
    current_price = (tick.bid + tick.ask) / 2
    forecast_price = current_price * (1 + price_change_pred)
    
    forecast_df = pd.DataFrame([{
        'forecast_price': forecast_price,
        'confidence': confidence,
        'current_price': current_price,
        'price_change_pred': price_change_pred,
        'timestamp': datetime.now()
    }])
    output_path = os.path.join(OUTPUT_DIR, f"forecast_{pair}.csv")
    forecast_df.to_csv(output_path, index=False)
    logger.info(f"Forecast for {pair} saved in {output_path}")
    
    visualize_forecast(pair, current_price, forecast_price, confidence)
    
    return {
        'pair': pair,
        'forecast_price': forecast_price,
        'confidence': max(0.0, min(1.0, confidence))
    }

結果の可視化

分析のために、現在価格と予測価格、およびネットCOTポジションとTFFポジションのチャートが作成されます。以下が可視化コードです。

def visualize_cot_data(cot_data: pd.DataFrame):
    if cot_data.empty or "Net_NonComm" not in cot_data.columns:
        logger.warning("No COT data available for visualization")
        return

    plt.figure(figsize=(14, 8))
    for market in cot_data["Market_and_Exchange_Names"].unique():
        market_data = cot_data[cot_data["Market_and_Exchange_Names"] == market]
        plt.plot(range(len(market_data)), market_data["Net_NonComm"], label=market, alpha=0.7)
    
    plt.title("Net Non-Commercial positions for currency futures")
    plt.xlabel("Entry")
    plt.ylabel("Net positions")
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    output_path = os.path.join(OUTPUT_DIR, "cot_net_positions.png")
    plt.savefig(output_path, dpi=150)
    plt.close()
    logger.info(f"Chart of net COT positions saved in {output_path}")

def visualize_tff_data(tff_data: pd.DataFrame):
    if tff_data.empty or "Net_Lev_Money" not in tff_data.columns:
        logger.warning("No TFF data available for visualization")
        return

    plt.figure(figsize=(14, 8))
    for market in tff_data["Market_and_Exchange_Names"].unique():
        market_data = tff_data[tff_data["Market_and_Exchange_Names"] == market]
        plt.plot(range(len(market_data)), market_data["Net_Lev_Money"], label=market, alpha=0.7)

    plt.title("Leveraged Funds net positions for currency futures (TFF)")
    plt.xlabel("Entry")
    plt.ylabel("Net positions")
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    output_path = os.path.join(OUTPUT_DIR, "tff_net_positions.png")
    plt.savefig(output_path, dpi=150)
    plt.close()
    logger.info(f"Chart of net TFF positions saved in {output_path}")

def visualize_forecast(pair: str, current_price: float, forecast_price: float, confidence: float):
    plt.figure(figsize=(8, 5))
    plt.bar(['Current Price', 'Forecast Price'], [current_price, forecast_price], color=['blue', 'green'], alpha=0.7)
    plt.title(f"Price forecast in 24 hours for {pair} (Confidence: {confidence:.2f})")
    plt.ylabel('Price')
    plt.tight_layout()
    output_path = os.path.join(OUTPUT_DIR, f"forecast_{pair}_plot.png")
    plt.savefig(output_path, dpi=150)
    plt.close()
    logger.info(f"Forecast graph saved to {output_path}")

これらのグラフにより、トレーダーは市場センチメントや予測精度を視覚的に把握できます。

最適化とエラー処理

データキャッシュ

ネットワークリクエストを最小限に抑えるため、COTおよびTFFデータはCSVファイルにキャッシュされ、最新データの再利用と高速な処理が可能になります。キャッシュはデータ読み込み前に確認され、存在しない場合または期限切れの場合にのみ新規リクエストが実行されます。

依存関係の確認

実行前に、xlrdおよびopenpyxlライブラリがインストールされているかを確認します。

def check_dependencies():
    dependencies = ['xlrd', 'openpyxl']
    for dep in dependencies:
        if not importlib.util.find_spec(dep):
            logger.error(f"Missing dependency: {dep}. Install it using 'pip install {dep}'.")
            raise ImportError(f"Missing dependency: {dep}")

エラー処理

このコードには、データの読み込み、MetaTrader 5への接続、およびモデルの学習といったすべての重要な処理に対する例外処理が含まれています。これにより、データ欠損や接続問題などの障害が発生した場合でもシステムの耐障害性が確保されます。

実用例

モデルのテスト

モデルを評価するためには、2024年の過去データ上でテストをおこなうことが推奨されます。R²指標および特徴量の重要度はCSVファイルに保存され、分析に使用されます。テストをおこなうことで、予測の精度を評価し、投機筋のネットポジションやボラティリティなどの最も重要な特徴量を特定することができます。

取引戦略への統合

予測結果はMetaTrader 5ライブラリを通じて自動売買に利用することができます。たとえば、予測された価格変化が0.5％を超え、かつモデルの信頼度が高い（0.7以上）の場合、ロングまたはショートポジションを開くことができます。以下が自動売買コードです。

def execute_trade(pair: str, forecast_price: float, confidence: float):
    if not mt5.initialize():
        logger.error("MT5 initialization failed")
        return

    symbol_info = mt5.symbol_info(pair)
    if not symbol_info:
        logger.warning(f"{pair} symbol not found")
        return

    current_price = (mt5.symbol_info_tick(pair).bid + mt5.symbol_info_tick(pair).ask) / 2
    if confidence > 0.7 and forecast_price > current_price * 1.005:
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": pair,
            "volume": 0.1,
            "type": mt5.ORDER_TYPE_BUY,
            "price": mt5.symbol_info_tick(pair).ask,
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_IOC,
        }
        result = mt5.order_send(request)
        logger.info(f"Opened long position for {pair}: {result}")
    elif confidence > 0.7 and forecast_price < current_price * 0.995:
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": pair,
            "volume": 0.1,
            "type": mt5.ORDER_TYPE_SELL,
            "price": mt5.symbol_info_tick(pair).bid,
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_IOC,
        }
        result = mt5.order_send(request)
        logger.info(f"Opened short position for {pair}: {result}")

機能拡張

モデルを改善するために、RSIやMACDなどのテクニカル指標を追加し、追加の市場シグナルを考慮することができます。XGBoostのような勾配ブースティングやニューラルネットワークなどの代替アルゴリズムを使用することで、予測精度を向上させることができます。COTおよびTFFデータの動的更新は、scheduleライブラリのようなタスクスケジューラを通じて実装でき、レポートが公開されるたびに自動的に新しいデータを取得することが可能です。また、特徴量を適切に調整することで、コモディティや指数など他の資産への拡張も可能です。

完全な実装：CurrencyForecastModule

統合を容易にするため、すべての機能はCurrencyForecastModuleクラスにまとめられており、データ読み込み、モデル学習、予測、および可視化をカプセル化しています。

class CurrencyForecastModule:
    def __init__(self, pairs: list, days_history: int = 30):
        self.pairs = pairs
        self.days_history = days_history
        self.models = {}
        self.scalers = {}
        self.forecasts = {}

        check_dependencies()
        if not mt5.initialize():
            logger.error("MT5 initialization failed. Ensure MT5 terminal is running and connected.")
            raise RuntimeError("MT5 initialization failed")

        self._validate_symbols()
        self._initialize_data()

    def _validate_symbols(self):
        available_symbols = [s.name for s in mt5.symbols_get()]
        logger.info(f"Available symbols in MT5: {available_symbols}")
        self.symbol_mapping = {}
        for pair in self.pairs[:]:
            if pair in available_symbols:
                self.symbol_mapping[pair] = pair
            else:
                base_pair = pair.split('.')[0]
                if base_pair in available_symbols:
                    self.symbol_mapping[pair] = base_pair
                    logger.info(f"Matched: {pair} -> {base_pair}")
                else:
                    logger.warning(f"{pair} symbol not found in MT5. Skipped.")
                    self.pairs.remove(pair)

    def _initialize_data(self):
        logger.info("Initializing data for CurrencyForecastModule...")
        self.cot_data = load_cot_reports()
        self.tff_data = self._load_tff_reports()
        for pair in self.pairs:
            self._train_model(pair)

    def _load_tff_reports(self) -> pd.DataFrame:
        cache_path = os.path.join(OUTPUT_DIR, "tff_report.csv")
        if os.path.exists(cache_path):
            logger.info(f"Loading TFF data from cache: {cache_path}")
            return pd.read_csv(cache_path)

        try:
            response = requests.get(TFF_URL)
            response.raise_for_status()
            zip_path = os.path.join(OUTPUT_DIR, "tff_data.zip")
            with open(zip_path, "wb") as f:
                f.write(response.content)

            with zipfile.ZipFile(zip_path, 'r') as zip_ref:
                zip_ref.extractall(OUTPUT_DIR)
                logger.info(f"TFF archive contents: {zip_ref.namelist()}")

            excel_files = glob.glob(os.path.join(OUTPUT_DIR, "**", "*.xls*"), recursive=True)
            tff_files = [f for f in excel_files if 'FinFut' in f or 'fin' in f.lower()]
            if not tff_files:
                logger.error("TFF Excel file not found in extracted files")
                return pd.DataFrame()

            excel_file = tff_files[0]
            logger.info(f"Handling TFF file: {excel_file}")

            relevant_columns = [
                "Market_and_Exchange_Names",
                "Lev_Money_Positions_Long_All",
                "Lev_Money_Positions_Short_All",
                "Asset_Mgr_Positions_Long_All",
                "Asset_Mgr_Positions_Short_All",
                "Open_Interest_All"
            ]
            tff_data = pd.read_excel(excel_file, engine='xlrd' if excel_file.endswith('.xls') else 'openpyxl')
            available_columns = [col for col in relevant_columns if col in tff_data.columns]
            if not available_columns:
                logger.error("Expected columns in TFF data not found")
                return pd.DataFrame()

            tff_data = tff_data[available_columns]
            forex_markets = ["EURO FX", "JAPANESE YEN", "BRITISH POUND", "AUSTRALIAN DOLLAR",
                             "CANADIAN DOLLAR", "SWISS FRANC", "MEXICAN PESO", "NEW ZEALAND DOLLAR"]
            tff_data = tff_data[tff_data["Market_and_Exchange_Names"].str.contains('|'.join(forex_markets), case=False, na=False)]
            if "Lev_Money_Positions_Long_All" in tff_data.columns:
                tff_data["Net_Lev_Money"] = tff_data["Lev_Money_Positions_Long_All"] - tff_data["Lev_Money_Positions_Short_All"]
            if "Asset_Mgr_Positions_Long_All" in tff_data.columns:
                tff_data["Net_Asset_Mgr"] = tff_data["Asset_Mgr_Positions_Long_All"] - tff_data["Asset_Mgr_Positions_Short_All"]

            tff_data.to_csv(cache_path, index=False)
            logger.info(f"TFF report saved in {cache_path}")
            visualize_tff_data(tff_data)
            return tff_data
        except Exception as e:
            logger.error(f"Error loading TFF data: {e}")
            return pd.DataFrame()

    def _train_model(self, pair: str):
        df = prepare_features(pair, self.cot_data, self.tff_data)
        model, scaler = train_model(pair, df)
        if model and scaler:
            self.models[pair] = model
            self.scalers[pair] = scaler

    async def update_forecasts(self):
        logger.info("Updating price forecasts...")
        for pair in self.pairs:
            forecast = await get_price_forecast(pair, self.models.get(pair), self.scalers.get(pair))
            logger.info(f"Forecast for {pair}: Price={forecast['forecast_price']}, Confidence={forecast['confidence']:.2f}")

    def __del__(self):
        mt5.shutdown()

このクラスは、データの読み込み、モデルの学習、および予測をおこなうためのモジュール構造を提供し、取引システムへの統合を簡素化します。

その結果、予測といくつかのグラフが得られます。ここでは例えば、COTとTFFポジションのチャートを示します。

最終的にこのコードは、複数銘柄に対して24時間の価格予測を生成します。

もちろん、より論理的にはCOTおよびTFFレポートの公開頻度と一致させるために、金曜日の終値を予測対象とする方が整合的です。 

結論

COTおよびTFFデータをMetaTrader 5のPythonライブラリを用いた過去価格データと組み合わせることで、取引戦略に組み込み可能な実用的な価格予測モデルを構築することができます。本実装はデータ読み込み、モデル学習、取引実行を自動化し、金融市場における分析および意思決定のための安定した基盤をトレーダーに提供します。このコードは新しい指標やアルゴリズムの追加によって拡張可能であり、またデータ更新の自動化により予測を常に最新の状態に保つことができます。