import time import json import logging # Optional: for better debugging # --- Configuration --- AUTH_TOKEN = "eyJhbGciOiJSUzUxMiIsImtpZCI6IkdaeFUiLCJ0eXAiOiJKV1QifQ.eyJ1c2VyX2lkIjoxMDQyNzg1ODYsImV4cCI6MTc0NDg2Njk4NSwiaWF0IjoxNzQ0ODUyNTg1LCJwbGFuIjoicHJvX3ByZW1pdW1fdHJpYWwiLCJkZWNsYXJlZF9zdGF0dXMiOiJub25fcHJvIiwiZXh0X2hvdXJzIjoxLCJwZXJtIjoiIiwic3R1ZHlfcGVybSI6InR2LWNoYXJ0X3BhdHRlcm5zLHR2LXByb3N0dWRpZXMsdHYtdm9sdW1lYnlwcmljZSx0di1jaGFydHBhdHRlcm5zIiwibWF4X3N0dWRpZXMiOjI1LCJtYXhfZnVuZGFtZW50YWxzIjoxMCwibWF4X2NoYXJ0cyI6OCwibWF4X2FjdGl2ZV9hbGVydHMiOjQwMCwibWF4X3N0dWR5X29uX3N0dWR5IjoyNCwiZmllbGRzX3Blcm1pc3Npb25zIjpbInJlZmJvbmRzIl0sIm1heF9vdmVyYWxsX2FsZXJ0cyI6MjAwMCwibWF4X292ZXJhbGxfd2F0Y2hsaXN0X2FsZXJ0cyI6NSwibWF4X2FjdGl2ZV9wcmltaXRpdmVfYWxlcnRzIjo0MDAsIm1heF9hY3RpdmVfY29tcGxleF9hbGVydHMiOjQwMCwibWF4X2FjdGl2ZV93YXRjaGxpc3RfYWxlcnRzIjoyLCJtYXhfY29ubmVjdGlvbnMiOjUwfQ.vo8X9P5XMSnpO9xob9vyieqUJJ5YIF3ypkQlIJV4UvWDFyULs7xPABFFp28RdAW17ZJPF2J7l6rVpp9ooYmTCZ2Yp5SXusxa64bsiVX4vEpE14B-zJUthxzrirqau8Nwtf8UEPZZGkQ2t_8yY_lsXt1M-2Tgp82mlHvh2NpAw0M" SYMBOL = "OANDA:XAUUSD" RESOLUTION = "1" # 1 minute # Example start time from logs (December 8, 2019 00:00:00 GMT approx) # You'll likely want to adjust this REPLAY_START_TIMESTAMP = 1575763199 REQUEST_CHUNK_SIZE = 100 # How many candles to request per chunk MAX_CHUNKS_TO_FETCH = 5 # Limit requests for this POC # --- State Variables --- chart_session_id = None replay_session_id = None replay_symbol_id = None chart_series_id = "sds_1" # Assuming we use the first series series_on_chart_id = "s2" # The ID used in modify_series, can be anything unique request_count = 0 all_candles = [] # --- Placeholder for your WebSocket client --- class MockWebSocketClient: def __init__(self): self.listeners = {} logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def send(self, event_name, params): message = {"m": event_name, "p": params} logging.info(f"SENDING: {json.dumps(message)}") # In a real scenario, this sends data over the WebSocket print(f"-> {json.dumps(message)}") # --- Simulate server responses for POC --- self.simulate_server_response(event_name, params) def add_listener(self, event_name, callback): if event_name not in self.listeners: self.listeners[event_name] = [] self.listeners[event_name].append(callback) logging.info(f"Listener added for: {event_name}") def on_message(self, message_str): logging.info(f"RECEIVED: {message_str}") print(f"<- {message_str}") try: message_obj = json.loads(message_str) # Handle potential list of messages if isinstance(message_obj, list): for msg in message_obj: self._dispatch_message(msg) elif isinstance(message_obj, dict) and 'm' in message_obj: self._dispatch_message(message_obj) else: logging.warning(f"Received non-standard message format: {message_str}") except json.JSONDecodeError: logging.error(f"Failed to decode JSON: {message_str}") except Exception as e: logging.error(f"Error processing message: {message_str} - {e}") def _dispatch_message(self, msg): event_name = msg.get('m') params = msg.get('p', []) if event_name in self.listeners: for callback in self.listeners[event_name]: try: callback(*params) # Pass parameters directly except Exception as e: logging.error(f"Error in callback for {event_name}: {e}") else: logging.debug(f"No listener for message: {event_name}") def simulate_server_response(self, sent_event, sent_params): # Simulate server responses based on sent messages - VERY BASIC global chart_session_id, replay_session_id, replay_symbol_id if sent_event == "set_auth_token": # Simulate successful auth (in reality, no specific success msg) # Then trigger the next step time.sleep(0.1) client.send("chart_create_session", ["cs_poc_123", ""]) elif sent_event == "chart_create_session": chart_session_id = sent_params[0] # Simulate success (again, no specific message, rely on next action) time.sleep(0.1) client.send("replay_create_session", ["rs_poc_456"]) elif sent_event == "replay_create_session": replay_session_id = sent_params[0] # Simulate instance ID response time.sleep(0.1) self.on_message(json.dumps({"m": "replay_instance_id", "p": [replay_session_id, "sim_server_1"]})) # Trigger next step time.sleep(0.1) client.send("replay_reset", [replay_session_id, "rp_reset_1", REPLAY_START_TIMESTAMP]) elif sent_event == "replay_reset": # Simulate replay_ok and replay_point time.sleep(0.1) self.on_message(json.dumps({"m": "replay_ok", "p": [sent_params[0], sent_params[1]]})) # replay_id, reset_req_id time.sleep(0.1) self.on_message(json.dumps({"m": "replay_point", "p": [sent_params[0], sent_params[2]]})) # replay_id, timestamp # Trigger next step time.sleep(0.1) symbol_str = f"={json.dumps({'symbol': SYMBOL, 'adjustment': 'splits', 'session': 'regular'})}" client.send("replay_add_series", [replay_session_id, "rp_series_1", symbol_str, RESOLUTION]) elif sent_event == "replay_add_series": # Simulate replay_ok and replay_resolutions time.sleep(0.1) self.on_message(json.dumps({"m": "replay_ok", "p": [sent_params[0], sent_params[1]]})) # replay_id, add_series_req_id time.sleep(0.1) self.on_message(json.dumps({"m": "replay_resolutions", "p": [sent_params[0], RESOLUTION, "1S"]})) # replay_id, resolution, min_resolution # Trigger next step time.sleep(0.1) replay_symbol_def = f"={json.dumps({'replay': replay_session_id, 'symbol': {'symbol': SYMBOL, 'adjustment': 'splits', 'session': 'regular', 'currency-id': 'USD'}})}" client.send("resolve_symbol", [chart_session_id, "sym_replay_1", replay_symbol_def]) elif sent_event == "resolve_symbol" and 'replay' in sent_params[2]: replay_symbol_id = sent_params[1] # Capture the ID sym_replay_1 # Simulate symbol_resolved time.sleep(0.1) self.on_message(json.dumps({"m": "symbol_resolved", "p": [chart_session_id, replay_symbol_id, {"name": SYMBOL, "pro_name": SYMBOL}]})) # simplified details # Trigger next step time.sleep(0.1) client.send("modify_series", [chart_session_id, chart_series_id, series_on_chart_id, replay_symbol_id, RESOLUTION, ""]) elif sent_event == "modify_series": # Simulate the server starting to send data automatically after modification time.sleep(0.2) self.simulate_timescale_update(is_initial=True) elif sent_event == "request_more_data": # Simulate getting more data time.sleep(0.3) self.simulate_timescale_update(is_initial=False) def simulate_timescale_update(self, is_initial): global request_count # Simulate receiving some candle data mock_candles = [] base_time = REPLAY_START_TIMESTAMP - (request_count * REQUEST_CHUNK_SIZE * 60) # Go back in time for i in range(REQUEST_CHUNK_SIZE): ts = base_time - (i * 60) # IMPORTANT: Real 'v' array in timescale_update seems to be [close, volume]. # OHLC needs separate handling or different requests if available. # For POC, we just simulate receiving *something*. # The actual structure from logs: {'i': index, 'v': [close, volume]} # We'll store timestamp, close, volume for this POC. close_price = 1500 + (request_count * 10) + i * 0.1 volume = 10 + i # Using the format seen in logs for timescale_update content: mock_candles.append({'i': -i - (request_count * REQUEST_CHUNK_SIZE), 'v': [close_price, volume]}) # Simulate timescale_update message structure update_msg = { "m": "timescale_update", "p": [ chart_session_id, { chart_series_id: { # Actually uses the 's2' ID here in logs after modify "s": mock_candles, "t": series_on_chart_id # Use the ID from modify_series } }, { "changes": [base_time - ((REQUEST_CHUNK_SIZE - 1) * 60)] if mock_candles else [], # Timestamp of first candle in this chunk "index": mock_candles[0]['i'] if mock_candles else 0 } ] } loading_msg = {"m": "series_loading", "p": [chart_session_id, chart_series_id, series_on_chart_id]} data_completed_status = {} if request_count >= MAX_CHUNKS_TO_FETCH -1 : data_completed_status = {"data_completed": "limit"} completed_msg = {"m": "series_completed", "p": [chart_session_id, chart_series_id, "replay", series_on_chart_id, data_completed_status]} self.on_message(json.dumps(loading_msg)) time.sleep(0.05) self.on_message(json.dumps(update_msg)) time.sleep(0.05) self.on_message(json.dumps(completed_msg)) client = MockWebSocketClient() # Replace with your actual client initialization # --- Callback Functions --- def on_connect(): # Assuming your library calls this on connection print("Connected!") client.send("set_auth_token", [AUTH_TOKEN]) # The rest of the flow is triggered by simulated responses in this POC def on_timescale_update(cs_id, data, time_info): global all_candles print("--- Received Candle Data Chunk ---") if chart_series_id in data: series_data = data[chart_series_id] elif series_on_chart_id in data: # Check the ID used in modify_series series_data = data[series_on_chart_id] else: print("Warning: Received timescale_update for unexpected series ID") return candles_in_chunk = series_data.get('s', []) # VERY Basic timestamp reconstruction (assumes 1-min interval exactly) first_ts = time_info.get('changes', [None])[0] first_idx = time_info.get('index', 0) print(f"Chunk Details: First TS: {first_ts}, First Idx: {first_idx}, Candles: {len(candles_in_chunk)}") reconstructed_candles = [] if first_ts is not None and candles_in_chunk: candles_in_chunk.sort(key=lambda x: x['i']) # Ensure order by index for candle in candles_in_chunk: index = candle['i'] # This timestamp calculation might need refinement based on real data timestamp = first_ts + (index - first_idx) * 60 close_price = candle['v'][0] volume = candle['v'][1] # We don't have OHLC from this message type easily reconstructed_candles.append({ "timestamp": timestamp, "close": close_price, "volume": volume, "index": index # Keep index for debugging }) print(f" - Index: {index}, Timestamp: {timestamp}, Close: {close_price}, Volume: {volume}") all_candles.extend(reconstructed_candles) print(f"Total candles received so far: {len(all_candles)}") print("---------------------------------") def on_series_completed(cs_id, series_id_internal, status, series_id_chart, details): global request_count print(f"Series Completed: InternalID={series_id_internal}, ChartID={series_id_chart}, Status={status}, Details={details}") # Check if we should request more data data_completed = details.get("data_completed") if status == "replay" and data_completed != "limit" and request_count < MAX_CHUNKS_TO_FETCH: print(f"Requesting next chunk ({request_count + 1}/{MAX_CHUNKS_TO_FETCH})...") time.sleep(0.5) # Small delay before next request request_count += 1 client.send("request_more_data", [chart_session_id, chart_series_id, REQUEST_CHUNK_SIZE]) elif data_completed == "limit": print("Received 'limit' signal. Stopping data requests.") print(f"\n--- FINAL CANDLE COUNT: {len(all_candles)} ---") # You might want to sort all_candles by timestamp here # for candle in sorted(all_candles, key=lambda x: x['timestamp']): # print(candle) elif request_count >= MAX_CHUNKS_TO_FETCH: print("Reached max request limit for POC. Stopping.") print(f"\n--- FINAL CANDLE COUNT: {len(all_candles)} ---") def on_replay_ok(rs_id, req_id): print(f"Replay OK received for session {rs_id}, request {req_id}") # No specific action needed here usually, handled by the flow in simulate_server_response def on_symbol_resolved(cs_id, symbol_req_id, details): print(f"Symbol resolved for request {symbol_req_id}: {details.get('pro_name')}") # No specific action needed here usually, handled by the flow # --- Register Listeners --- client.add_listener("timescale_update", on_timescale_update) client.add_listener("series_completed", on_series_completed) client.add_listener("replay_ok", on_replay_ok) # Useful for debugging client.add_listener("symbol_resolved", on_symbol_resolved) # Useful for debugging # --- Start the process --- # In a real client, you'd connect first, and 'on_connect' would trigger the chain. # For this POC, we simulate the connection and first step. print("Starting POC...") on_connect() # Keep the script running to receive messages (in a real app, this would be an event loop) print("POC running simulation... (will stop after MAX_CHUNKS_TO_FETCH)") # Wait long enough for the simulation to run timeout = (MAX_CHUNKS_TO_FETCH + 2) * 1.5 # Estimate time based on sleeps and chunks time.sleep(timeout) print("POC finished.")