import json import datetime from ib_insync import IB, Future, util def get_third_friday(year, month): """ Returns the third Friday of the given year and month as a datetime, which is a common expiration for futures. """ fridays = [] for day in range(1, 32): try: d = datetime.date(year, month, day) except ValueError: break if d.weekday() == 4: # Friday fridays.append(d) if len(fridays) >= 3: return datetime.datetime.combine(fridays[2], datetime.time(16, 0)) # 4:00 PM elif fridays: return datetime.datetime.combine(fridays[-1], datetime.time(16, 0)) else: return datetime.datetime(year, month, 1, 16, 0) def generate_contract_months(start_date, end_date): """ Generate a sorted list of contract month strings (format "YYYYMM") that might be active between start_date and end_date. MES futures are listed for quarters (Mar, Jun, Sep, Dec). """ months = [3, 6, 9, 12] result = [] for year in range(start_date.year, end_date.year + 2): for m in months: dt = datetime.datetime(year, m, 1) if dt <= end_date: result.append(f"{year}{m:02d}") return sorted(set(result)) def get_data_chunk(contract, end_dt, duration_str="1 W"): """ Request a chunk of historical data for the given contract ending at end_dt. Returns a list of bars (or None on error). """ try: bars = ib.reqHistoricalData( contract, endDateTime=end_dt.strftime("%Y%m%d %H:%M:%S"), durationStr=duration_str, barSizeSetting="5 mins", whatToShow="TRADES", useRTH=True, # Regular trading hours only formatDate=1 ) return bars except Exception as e: print(f"Error retrieving data chunk for {contract.localSymbol} ending at {end_dt}: {e}") return None def getMESContract(cm, contract_expiration): """ Attempt multiple MES contract definitions for the given contract month (cm) and expiration date. Returns the first valid contract (as returned by reqContractDetails) or None. """ expiration_str = contract_expiration.strftime("%Y%m%d") variants = [] # Variant 1: Use full expiration date (YYYYMMDD), exchange GLOBEX, minimal fields. contract1 = Future( symbol='MES', lastTradeDateOrContractMonth=expiration_str, exchange='GLOBEX', currency='USD', multiplier=5 ) contract1.includeExpired = True variants.append(("Variant 1: full expiration date, GLOBEX", contract1)) # Variant 2: Full expiration date, exchange CME. contract2 = Future( symbol='MES', lastTradeDateOrContractMonth=expiration_str, exchange='CME', currency='USD', multiplier=5 ) contract2.includeExpired = True variants.append(("Variant 2: full expiration date, CME", contract2)) # Variant 3: Use contract month (YYYYMM), exchange GLOBEX, add tradingClass. contract3 = Future( symbol='MES', lastTradeDateOrContractMonth=cm, exchange='GLOBEX', currency='USD', multiplier=5, tradingClass='MES' ) contract3.includeExpired = True variants.append(("Variant 3: contract month, GLOBEX, tradingClass MES", contract3)) # Variant 4: Use contract month, exchange CME, add tradingClass. contract4 = Future( symbol='MES', lastTradeDateOrContractMonth=cm, exchange='CME', currency='USD', multiplier=5, tradingClass='MES' ) contract4.includeExpired = True variants.append(("Variant 4: contract month, CME, tradingClass MES", contract4)) # Variant 5: Use contract month, exchange GLOBEX, with a computed localSymbol. month_codes = {1:'F', 2:'G', 3:'H', 4:'J', 5:'K', 6:'M', 7:'N', 8:'Q', 9:'U', 10:'V', 11:'X', 12:'Z'} year = int(cm[:4]) month = int(cm[4:]) local_symbol = f"MES{month_codes.get(month, '')}{str(year)[-1]}" contract5 = Future( symbol='MES', lastTradeDateOrContractMonth=cm, localSymbol=local_symbol, exchange='GLOBEX', currency='USD', multiplier=5 ) contract5.includeExpired = True variants.append(("Variant 5: contract month, GLOBEX, localSymbol", contract5)) # Try each variant. for variant_desc, contract in variants: print(f"Trying {variant_desc} for contract month {cm} (expiration: {expiration_str})...") details = ib.reqContractDetails(contract) if details: print(f"Success with {variant_desc}: found contract details: {details[0].contract}") return details[0].contract else: print(f"{variant_desc} did not return contract details.") return None # --- Main Script --- # Connect to IB Gateway (ensure your account is active and market data is subscribed) ib = IB() try: ib.connect('127.0.0.1', 4002, clientId=1) except Exception as e: print(f"Connection error: {e}") exit(1) # Define overall desired time range: last 3 years up until today. # We'll use naive datetime objects (local time) end_date = datetime.datetime.now() start_date = end_date - datetime.timedelta(days=3*365) # Generate contract month strings (e.g., "202303", "202306", etc.) contract_months = generate_contract_months(start_date, end_date) print("Contract months to process:", contract_months) all_data = [] # Process each contract month for cm in contract_months: year = int(cm[:4]) month = int(cm[4:]) # Compute the contract expiration date (using third Friday) contract_expiration = get_third_friday(year, month) # Try to obtain a valid MES contract using our diagnostic function. mes_contract = getMESContract(cm, contract_expiration) if not mes_contract: print(f"*** No valid MES contract found for {cm}. Skipping this month. ***") continue print(f"Processing contract {mes_contract.localSymbol} (approx expiration: {contract_expiration})") # Determine the effective data period for this contract. contract_end = min(end_date, contract_expiration) current_end = contract_end contract_data = [] chunk_duration = datetime.timedelta(weeks=1) # Request data in weekly chunks until we reach start_date. while current_end > start_date: print(f" Requesting {mes_contract.localSymbol} data ending at {current_end}") bars = get_data_chunk(mes_contract, current_end, duration_str="1 W") if bars is None: print(" Error retrieving chunk; moving to next week.") current_end -= chunk_duration continue if len(bars) == 0: print(" No data returned for this period; ending requests for this contract.") break for bar in bars: # Remove timezone info from bar.date so we can compare with naive datetimes. bar_time = bar.date.replace(tzinfo=None) if start_date <= bar_time <= end_date: contract_data.append({ 'date': bar_time.strftime("%Y-%m-%d %H:%M:%S"), 'open': bar.open, 'high': bar.high, 'low': bar.low, 'close': bar.close, 'volume': bar.volume, 'contract': mes_contract.localSymbol }) earliest_time = min(bar.date.replace(tzinfo=None) for bar in bars) new_end = earliest_time - datetime.timedelta(seconds=1) if new_end >= current_end: break current_end = new_end print(f" Retrieved {len(contract_data)} bars for contract {mes_contract.localSymbol}") all_data.extend(contract_data) # Remove duplicate bars (if any) based on timestamp. unique_data = {d['date']: d for d in all_data} final_data = sorted(unique_data.values(), key=lambda x: x['date']) expected_bars = ((end_date - start_date).total_seconds() / (5 * 60)) if len(final_data) < expected_bars * 0.9: print("Warning: Retrieved data appears significantly less than expected.") output_filename = "mes_5min_data.json" if final_data: try: with open(output_filename, "w") as f: json.dump(final_data, f, indent=4) print(f"Data successfully saved to {output_filename}") except Exception as e: print(f"Error writing to JSON file: {e}") else: print("No data retrieved. File not saved.") ib.disconnect()