Commit inicial: Projeto YuClusters
This commit is contained in:
Thiago Moura
@@ -0,0 +1,389 @@
|
||||
import uuid
|
||||
import time
|
||||
from typing import Dict, Any, List, Optional
|
||||
from config import settings
|
||||
|
||||
class FootprintCluster:
|
||||
def __init__(self, tick_size: float = 1.0):
|
||||
self.tick_size = tick_size
|
||||
self.cluster_id = str(uuid.uuid4())
|
||||
self.status = "active"
|
||||
self.open_time: Optional[int] = None
|
||||
self.close_reason: Optional[str] = None
|
||||
self.open_price: Optional[float] = None
|
||||
self.close_price: Optional[float] = None
|
||||
self.high: Optional[float] = None
|
||||
self.low: Optional[float] = None
|
||||
self.poc: Optional[float] = None
|
||||
self.total_delta: float = 0.0
|
||||
self.total_volume: float = 0.0
|
||||
|
||||
# levels: price_float -> { 'ask': float, 'bid': float }
|
||||
# internally we keep float keys to make sorting and arithmetic easy
|
||||
self.levels: Dict[float, Dict[str, float]] = {}
|
||||
self.stacked: Dict[str, Any] = {
|
||||
"buy": False,
|
||||
"sell": False,
|
||||
"price_range": []
|
||||
}
|
||||
self.advanced_metrics: Dict[str, Any] = {
|
||||
"poc_position": None,
|
||||
"pattern": None,
|
||||
"top_extreme": None,
|
||||
"bottom_extreme": None
|
||||
}
|
||||
|
||||
def add_tick(self, price: float, volume: float, is_buy: bool, timestamp_msc: int) -> None:
|
||||
if self.open_time is None:
|
||||
self.open_time = timestamp_msc
|
||||
self.open_price = price
|
||||
|
||||
self.close_price = price
|
||||
|
||||
# Round price to the nearest tick_size to avoid float precision issues
|
||||
rounded_price = round(price / self.tick_size) * self.tick_size
|
||||
|
||||
if rounded_price not in self.levels:
|
||||
self.levels[rounded_price] = {"ask": 0.0, "bid": 0.0}
|
||||
|
||||
if is_buy:
|
||||
self.levels[rounded_price]["ask"] += volume
|
||||
else:
|
||||
self.levels[rounded_price]["bid"] += volume
|
||||
|
||||
# Update High/Low
|
||||
if self.high is None or rounded_price > self.high:
|
||||
self.high = rounded_price
|
||||
if self.low is None or rounded_price < self.low:
|
||||
self.low = rounded_price
|
||||
|
||||
# Recalculate totals, POC, imbalances and stacked imbalances
|
||||
self._recalculate()
|
||||
|
||||
def _recalculate(self) -> None:
|
||||
if not self.levels:
|
||||
return
|
||||
|
||||
self.total_volume = 0.0
|
||||
self.total_delta = 0.0
|
||||
|
||||
# First pass: calc totals and delta per level
|
||||
for price, data in self.levels.items():
|
||||
ask = data["ask"]
|
||||
bid = data["bid"]
|
||||
lvl_delta = ask - bid
|
||||
lvl_total = ask + bid
|
||||
self.total_volume += lvl_total
|
||||
self.total_delta += lvl_delta
|
||||
|
||||
# Find POC: level with the highest total volume.
|
||||
# Tie-breaker: choose the highest price level.
|
||||
sorted_prices = sorted(self.levels.keys())
|
||||
best_price = sorted_prices[0]
|
||||
max_total = -1.0
|
||||
for price in sorted_prices:
|
||||
total_vol = self.levels[price]["ask"] + self.levels[price]["bid"]
|
||||
if total_vol > max_total:
|
||||
max_total = total_vol
|
||||
best_price = price
|
||||
elif total_vol == max_total:
|
||||
if price > best_price:
|
||||
best_price = price
|
||||
self.poc = best_price
|
||||
|
||||
# Second pass: calculate imbalances diagonal and stacked imbalances
|
||||
# To avoid division by zero:
|
||||
# imbalance_buy[i] is True if ask_vol[i] >= R * bid_vol[i - tick_size]
|
||||
# imbalance_sell[i] is True if bid_vol[i] >= R * ask_vol[i + tick_size]
|
||||
R = settings.IMBALANCE_RATIO
|
||||
|
||||
imbalances_buy = {}
|
||||
imbalances_sell = {}
|
||||
|
||||
for price in sorted_prices:
|
||||
ask_val = self.levels[price]["ask"]
|
||||
bid_val = self.levels[price]["bid"]
|
||||
|
||||
# Lower level price
|
||||
lower_price = round((price - self.tick_size) / self.tick_size) * self.tick_size
|
||||
bid_below = self.levels[lower_price]["bid"] if lower_price in self.levels else 0.0
|
||||
|
||||
# Upper level price
|
||||
upper_price = round((price + self.tick_size) / self.tick_size) * self.tick_size
|
||||
ask_above = self.levels[upper_price]["ask"] if upper_price in self.levels else 0.0
|
||||
|
||||
# Buy Imbalance (diagonal): ask vs bid_below
|
||||
# Avoid triggering on 0 vs 0
|
||||
if ask_val > 0 and ask_val >= R * bid_below:
|
||||
imbalances_buy[price] = True
|
||||
else:
|
||||
imbalances_buy[price] = False
|
||||
|
||||
# Sell Imbalance (diagonal): bid vs ask_above
|
||||
if bid_val > 0 and bid_val >= R * ask_above:
|
||||
imbalances_sell[price] = True
|
||||
else:
|
||||
imbalances_sell[price] = False
|
||||
|
||||
# Detect stacked imbalances: 3+ consecutive levels with imbalance in the same direction
|
||||
# Let's check contiguous price levels in steps of tick_size
|
||||
stacked_buy = False
|
||||
stacked_sell = False
|
||||
stacked_buy_prices = []
|
||||
stacked_sell_prices = []
|
||||
|
||||
min_consecutive = settings.STACKED_MIN_COUNT
|
||||
|
||||
# We need to check all possible price steps from low to high
|
||||
if self.low is not None and self.high is not None:
|
||||
current_price = self.low
|
||||
consec_buy = []
|
||||
consec_sell = []
|
||||
|
||||
while current_price <= self.high:
|
||||
rounded_p = round(current_price / self.tick_size) * self.tick_size
|
||||
|
||||
# Check Buy
|
||||
if imbalances_buy.get(rounded_p, False):
|
||||
consec_buy.append(rounded_p)
|
||||
else:
|
||||
if len(consec_buy) >= min_consecutive:
|
||||
stacked_buy = True
|
||||
stacked_buy_prices.extend(consec_buy)
|
||||
consec_buy = []
|
||||
|
||||
# Check Sell
|
||||
if imbalances_sell.get(rounded_p, False):
|
||||
consec_sell.append(rounded_p)
|
||||
else:
|
||||
if len(consec_sell) >= min_consecutive:
|
||||
stacked_sell = True
|
||||
stacked_sell_prices.extend(consec_sell)
|
||||
consec_sell = []
|
||||
|
||||
current_price += self.tick_size
|
||||
|
||||
# Final check at the end of the loop
|
||||
if len(consec_buy) >= min_consecutive:
|
||||
stacked_buy = True
|
||||
stacked_buy_prices.extend(consec_buy)
|
||||
if len(consec_sell) >= min_consecutive:
|
||||
stacked_sell = True
|
||||
stacked_sell_prices.extend(consec_sell)
|
||||
|
||||
# Build output properties for stacked
|
||||
self.stacked = {
|
||||
"buy": stacked_buy,
|
||||
"sell": stacked_sell,
|
||||
"price_range": sorted(list(set(stacked_buy_prices + stacked_sell_prices)))
|
||||
}
|
||||
|
||||
# Store imbalances back in levels for JSON output
|
||||
for price in sorted_prices:
|
||||
imb = None
|
||||
if imbalances_buy.get(price, False) and imbalances_sell.get(price, False):
|
||||
imb = "both"
|
||||
elif imbalances_buy.get(price, False):
|
||||
imb = "buy"
|
||||
elif imbalances_sell.get(price, False):
|
||||
imb = "sell"
|
||||
|
||||
self.levels[price]["delta"] = self.levels[price]["ask"] - self.levels[price]["bid"]
|
||||
self.levels[price]["total"] = self.levels[price]["ask"] + self.levels[price]["bid"]
|
||||
self.levels[price]["imbalance"] = imb
|
||||
|
||||
# Advanced Metrics Calculation
|
||||
if self.high is not None and self.low is not None and self.high > self.low and self.total_volume > 0:
|
||||
# 1. POC Position
|
||||
poc_percent = (self.poc - self.low) / (self.high - self.low)
|
||||
if poc_percent >= 0.65:
|
||||
self.advanced_metrics["poc_position"] = "top"
|
||||
elif poc_percent <= 0.35:
|
||||
self.advanced_metrics["poc_position"] = "bottom"
|
||||
else:
|
||||
self.advanced_metrics["poc_position"] = "middle"
|
||||
|
||||
# 2. P and B Patterns
|
||||
mid_price = (self.high + self.low) / 2.0
|
||||
vol_above = sum(self.levels[p]["total"] for p in sorted_prices if p >= mid_price)
|
||||
vol_below = sum(self.levels[p]["total"] for p in sorted_prices if p < mid_price)
|
||||
|
||||
if vol_above / self.total_volume > 0.65:
|
||||
self.advanced_metrics["pattern"] = "P"
|
||||
elif vol_below / self.total_volume > 0.65:
|
||||
self.advanced_metrics["pattern"] = "B"
|
||||
else:
|
||||
self.advanced_metrics["pattern"] = "normal"
|
||||
|
||||
# 3. Extremes (Exhaustion / Absorption)
|
||||
top_vol = self.levels[self.high]["total"]
|
||||
next_top_price = round((self.high - self.tick_size) / self.tick_size) * self.tick_size
|
||||
next_top_vol = self.levels.get(next_top_price, {}).get("total", 0.0)
|
||||
avg_vol = self.total_volume / len(self.levels)
|
||||
|
||||
if top_vol < avg_vol * 0.2 and next_top_vol > avg_vol * 0.5:
|
||||
self.advanced_metrics["top_extreme"] = "exhaustion"
|
||||
elif top_vol > avg_vol * 2.5:
|
||||
self.advanced_metrics["top_extreme"] = "absorption"
|
||||
else:
|
||||
self.advanced_metrics["top_extreme"] = "normal"
|
||||
|
||||
bottom_vol = self.levels[self.low]["total"]
|
||||
next_bot_price = round((self.low + self.tick_size) / self.tick_size) * self.tick_size
|
||||
next_bot_vol = self.levels.get(next_bot_price, {}).get("total", 0.0)
|
||||
|
||||
if bottom_vol < avg_vol * 0.2 and next_bot_vol > avg_vol * 0.5:
|
||||
self.advanced_metrics["bottom_extreme"] = "exhaustion"
|
||||
elif bottom_vol > avg_vol * 2.5:
|
||||
self.advanced_metrics["bottom_extreme"] = "absorption"
|
||||
else:
|
||||
self.advanced_metrics["bottom_extreme"] = "normal"
|
||||
|
||||
# 4. Ratios (Exhaustion/Absorption quantification at extremes)
|
||||
if next_top_vol > 0:
|
||||
self.advanced_metrics["high_ratio"] = round(top_vol / next_top_vol, 2)
|
||||
else:
|
||||
self.advanced_metrics["high_ratio"] = round(top_vol, 2) if top_vol > 0 else 0.0
|
||||
|
||||
if next_bot_vol > 0:
|
||||
self.advanced_metrics["low_ratio"] = round(bottom_vol / next_bot_vol, 2)
|
||||
else:
|
||||
self.advanced_metrics["low_ratio"] = round(bottom_vol, 2) if bottom_vol > 0 else 0.0
|
||||
|
||||
# 5. Delta Divergence (Price direction vs Order Flow Delta)
|
||||
self.advanced_metrics["delta_divergence"] = False
|
||||
if self.open_price is not None and self.close_price is not None:
|
||||
is_bull = self.close_price > self.open_price
|
||||
is_bear = self.close_price < self.open_price
|
||||
if (is_bull and self.total_delta < 0) or (is_bear and self.total_delta > 0):
|
||||
self.advanced_metrics["delta_divergence"] = True
|
||||
|
||||
def should_close(self, current_time_msc: int) -> Optional[str]:
|
||||
# Check range
|
||||
if self.high is not None and self.low is not None:
|
||||
points = self.high - self.low
|
||||
# If tick_size is 1.0, range is in points. Let's compare directly.
|
||||
# (or points/tick_size >= CLUSTER_RANGE_POINTS)
|
||||
# Standard MT5 points. If range in points exceeds setting:
|
||||
if points >= settings.CLUSTER_RANGE_POINTS:
|
||||
return "range"
|
||||
|
||||
# Check volume
|
||||
if self.total_volume >= settings.CLUSTER_VOLUME_MAX:
|
||||
return "volume"
|
||||
|
||||
# Check delta
|
||||
if abs(self.total_delta) >= settings.CLUSTER_DELTA_MAX:
|
||||
return "delta"
|
||||
|
||||
# Check time
|
||||
if self.open_time is not None:
|
||||
elapsed = (current_time_msc - self.open_time) / 1000.0
|
||||
if elapsed >= settings.CLUSTER_TIME_SECONDS:
|
||||
return "time"
|
||||
|
||||
return None
|
||||
|
||||
def close(self, reason: str) -> None:
|
||||
self.status = "closed"
|
||||
self.close_reason = reason
|
||||
|
||||
def to_json(self) -> Dict[str, Any]:
|
||||
# Convert keys in levels to string for JSON compatibility
|
||||
levels_str = {}
|
||||
for price, data in sorted(self.levels.items(), reverse=True):
|
||||
levels_str[f"{price:.5f}"] = {
|
||||
"ask": float(data.get("ask", 0.0)),
|
||||
"bid": float(data.get("bid", 0.0)),
|
||||
"delta": float(data.get("delta", 0.0)),
|
||||
"total": float(data.get("total", 0.0)),
|
||||
"imbalance": data.get("imbalance", None)
|
||||
}
|
||||
|
||||
return {
|
||||
"cluster_id": self.cluster_id,
|
||||
"tick_size": self.tick_size,
|
||||
"status": self.status,
|
||||
"open_time": int(self.open_time) if self.open_time is not None else None,
|
||||
"close_reason": self.close_reason,
|
||||
"open_price": float(self.open_price) if self.open_price is not None else None,
|
||||
"close_price": float(self.close_price) if self.close_price is not None else None,
|
||||
"high": float(self.high) if self.high is not None else None,
|
||||
"low": float(self.low) if self.low is not None else None,
|
||||
"poc": float(self.poc) if self.poc is not None else None,
|
||||
"total_delta": float(self.total_delta),
|
||||
"total_volume": float(self.total_volume),
|
||||
"levels": levels_str,
|
||||
"stacked": {
|
||||
"buy": bool(self.stacked.get("buy", False)),
|
||||
"sell": bool(self.stacked.get("sell", False)),
|
||||
"price_range": [float(p) for p in self.stacked.get("price_range", [])]
|
||||
},
|
||||
"advanced_metrics": self.advanced_metrics
|
||||
}
|
||||
|
||||
|
||||
class Aggregator:
|
||||
def __init__(self, tick_size: float = 1.0):
|
||||
self.tick_size = tick_size
|
||||
self.active_cluster = FootprintCluster(tick_size=self.tick_size)
|
||||
self.history: List[Dict[str, Any]] = []
|
||||
|
||||
def process_tick(self, price: float, volume: float, is_buy: bool, timestamp_msc: int) -> tuple[Dict[str, Any], Optional[Dict[str, Any]]]:
|
||||
"""
|
||||
Process a single tick.
|
||||
Returns:
|
||||
(active_cluster_json, closed_cluster_json_if_just_closed)
|
||||
"""
|
||||
self.active_cluster.add_tick(price, volume, is_buy, timestamp_msc)
|
||||
|
||||
# Check if the active cluster should be closed
|
||||
close_reason = self.active_cluster.should_close(timestamp_msc)
|
||||
|
||||
closed_json = None
|
||||
if close_reason:
|
||||
self.active_cluster.close(close_reason)
|
||||
closed_json = self.active_cluster.to_json()
|
||||
|
||||
# Save to history
|
||||
self.history.append(closed_json)
|
||||
if len(self.history) > settings.HISTORY_BUFFER_SIZE:
|
||||
self.history.pop(0)
|
||||
|
||||
# Start new cluster
|
||||
self.active_cluster = FootprintCluster(tick_size=self.tick_size)
|
||||
# The next tick will set the open_time of the new cluster.
|
||||
|
||||
return self.active_cluster.to_json(), closed_json
|
||||
|
||||
|
||||
def classify_tick(last: float, bid: float, ask: float, flags: int) -> bool:
|
||||
"""
|
||||
Classify tick aggressiveness.
|
||||
TICK_FLAG_BUY = 32 -> BUY (True)
|
||||
TICK_FLAG_SELL = 64 -> SELL (False)
|
||||
TICK_FLAG_ASK = 4 -> ASK (True)
|
||||
TICK_FLAG_BID = 2 -> BID (False)
|
||||
"""
|
||||
if flags & 32:
|
||||
return True
|
||||
elif flags & 64:
|
||||
return False
|
||||
|
||||
if last > 0:
|
||||
if last >= ask:
|
||||
return True
|
||||
elif last <= bid:
|
||||
return False
|
||||
|
||||
# Forex quote ticks fallback
|
||||
if flags & 4:
|
||||
return True
|
||||
elif flags & 2:
|
||||
return False
|
||||
|
||||
# Standard fallback if between spread: closer to ask is BUY
|
||||
if ask > bid and last > 0:
|
||||
return (last - bid) >= (ask - last)
|
||||
return True
|
||||
|
||||
@@ -0,0 +1,175 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
import sys
|
||||
import os
|
||||
from typing import Callable, Optional, Any
|
||||
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
||||
|
||||
import MetaTrader5 as mt5
|
||||
from config import settings
|
||||
from backend.aggregator import Aggregator, classify_tick
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||||
logger = logging.getLogger("mt5_collector")
|
||||
|
||||
class MT5Collector:
|
||||
def __init__(self, aggregator: Aggregator, on_update_callback: Callable[[dict, Optional[dict]], Any]):
|
||||
self.aggregator = aggregator
|
||||
self.on_update_callback = on_update_callback
|
||||
self.symbol = settings.MT5_SYMBOL
|
||||
self.running = False
|
||||
self.connected = False
|
||||
self.last_tick_time_msc = 0
|
||||
self.seen_ticks_buffer = set()
|
||||
|
||||
async def connect_mt5(self) -> bool:
|
||||
"""
|
||||
Attempts to initialize and login to the MetaTrader 5 terminal.
|
||||
All MT5 calls are blocking, so they run in a thread executor.
|
||||
"""
|
||||
try:
|
||||
mt5_path = r"C:\Program Files\MetaTrader 5\terminal64.exe"
|
||||
initialized = await asyncio.to_thread(mt5.initialize, path=mt5_path)
|
||||
if not initialized:
|
||||
err = await asyncio.to_thread(mt5.last_error)
|
||||
logger.error(f"MT5 initialize failed: {err}")
|
||||
return False
|
||||
|
||||
if settings.MT5_LOGIN > 0:
|
||||
login_success = await asyncio.to_thread(
|
||||
mt5.login,
|
||||
settings.MT5_LOGIN,
|
||||
password=settings.MT5_PASSWORD,
|
||||
server=settings.MT5_SERVER,
|
||||
)
|
||||
if not login_success:
|
||||
err = await asyncio.to_thread(mt5.last_error)
|
||||
logger.error(f"MT5 login failed: {err}")
|
||||
await asyncio.to_thread(mt5.shutdown)
|
||||
return False
|
||||
|
||||
symbol_info = await asyncio.to_thread(mt5.symbol_info, self.symbol)
|
||||
if symbol_info is None:
|
||||
logger.error(f"Symbol {self.symbol} not found.")
|
||||
await asyncio.to_thread(mt5.shutdown)
|
||||
return False
|
||||
|
||||
if not symbol_info.visible:
|
||||
selected = await asyncio.to_thread(mt5.symbol_select, self.symbol, True)
|
||||
if not selected:
|
||||
logger.error(f"Failed to select/make visible symbol {self.symbol}.")
|
||||
await asyncio.to_thread(mt5.shutdown)
|
||||
return False
|
||||
|
||||
tick_size = symbol_info.trade_tick_size
|
||||
if tick_size > 0:
|
||||
self.aggregator.tick_size = tick_size
|
||||
self.aggregator.active_cluster.tick_size = tick_size
|
||||
logger.info(f"Set aggregator tick size to {tick_size}")
|
||||
|
||||
logger.info("Successfully connected to MetaTrader 5 and logged in.")
|
||||
self.connected = True
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Exception during MT5 connection: {e}")
|
||||
return False
|
||||
|
||||
async def disconnect_mt5(self):
|
||||
try:
|
||||
await asyncio.to_thread(mt5.shutdown)
|
||||
except Exception as e:
|
||||
logger.error(f"Error during MT5 shutdown: {e}")
|
||||
self.connected = False
|
||||
|
||||
async def start(self):
|
||||
self.running = True
|
||||
backoff = 1.0
|
||||
|
||||
while self.running:
|
||||
if not self.connected:
|
||||
success = await self.connect_mt5()
|
||||
if not success:
|
||||
logger.info(f"Reconnecting to MT5 in {backoff:.1f}s...")
|
||||
await asyncio.sleep(backoff)
|
||||
backoff = min(backoff * 2, 60.0)
|
||||
continue
|
||||
else:
|
||||
backoff = 1.0
|
||||
|
||||
# Fetch from 6 hours ago so the chart isn't empty when started
|
||||
from datetime import datetime, timedelta
|
||||
start_time_dt = datetime.now() - timedelta(hours=6)
|
||||
ticks = await asyncio.to_thread(
|
||||
mt5.copy_ticks_from, self.symbol, start_time_dt, 100000, mt5.COPY_TICKS_ALL
|
||||
)
|
||||
if ticks is not None and len(ticks) > 0:
|
||||
self.last_tick_time_msc = ticks[0]['time_msc']
|
||||
else:
|
||||
self.last_tick_time_msc = int(start_time_dt.timestamp() * 1000)
|
||||
|
||||
# Polling loop
|
||||
try:
|
||||
from datetime import datetime
|
||||
polling_dt = datetime.fromtimestamp(self.last_tick_time_msc / 1000.0)
|
||||
ticks = await asyncio.to_thread(
|
||||
mt5.copy_ticks_from,
|
||||
self.symbol,
|
||||
polling_dt,
|
||||
1000,
|
||||
mt5.COPY_TICKS_ALL,
|
||||
)
|
||||
|
||||
if ticks is None:
|
||||
err = await asyncio.to_thread(mt5.last_error)
|
||||
logger.error(f"MT5 copy_ticks_from returned None: {err}")
|
||||
self.connected = False
|
||||
await self.disconnect_mt5()
|
||||
continue
|
||||
|
||||
if len(ticks) > 0:
|
||||
logger.info(f"Fetched {len(ticks)} ticks starting at {ticks[0]['time_msc']}")
|
||||
for tick in ticks:
|
||||
msc = tick['time_msc']
|
||||
|
||||
if msc < self.last_tick_time_msc:
|
||||
continue
|
||||
|
||||
tick_id = (msc, tick['bid'], tick['ask'], tick['last'], tick['volume_real'], tick['flags'])
|
||||
if msc == self.last_tick_time_msc and tick_id in self.seen_ticks_buffer:
|
||||
continue
|
||||
|
||||
if msc > self.last_tick_time_msc:
|
||||
self.seen_ticks_buffer.clear()
|
||||
self.last_tick_time_msc = msc
|
||||
|
||||
self.seen_ticks_buffer.add(tick_id)
|
||||
|
||||
price = tick['last'] if tick['last'] > 0 else (tick['bid'] if tick['bid'] > 0 else tick['ask'])
|
||||
volume = tick['volume_real'] if tick['volume_real'] > 0 else float(tick['volume'])
|
||||
flags = tick['flags']
|
||||
|
||||
# Forex ticks often have volume=0 (they are just quote updates).
|
||||
# We count each quote update as 1 unit of tick volume to build the footprint.
|
||||
if volume == 0:
|
||||
volume = 1.0
|
||||
|
||||
is_buy = classify_tick(price, tick['bid'], tick['ask'], flags)
|
||||
|
||||
active_json, closed_json = self.aggregator.process_tick(price, volume, is_buy, msc)
|
||||
|
||||
self.on_update_callback(active_json, closed_json)
|
||||
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error during tick polling loop: {e}")
|
||||
self.connected = False
|
||||
await self.disconnect_mt5()
|
||||
await asyncio.sleep(2.0)
|
||||
|
||||
async def stop(self):
|
||||
self.running = False
|
||||
await self.disconnect_mt5()
|
||||
logger.info("MT5 Collector stopped.")
|
||||
@@ -0,0 +1,116 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import sys
|
||||
import os
|
||||
from contextlib import asynccontextmanager
|
||||
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from typing import Set, Dict, Any, Optional
|
||||
|
||||
# Add project root to sys.path
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
||||
|
||||
from config import settings
|
||||
from backend.aggregator import Aggregator
|
||||
from backend.mt5_collector import MT5Collector
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||||
logger = logging.getLogger("server")
|
||||
|
||||
# Shared state
|
||||
aggregator = Aggregator(tick_size=1.0)
|
||||
active_connections: Set[WebSocket] = set()
|
||||
collector_task: Optional[asyncio.Task] = None
|
||||
|
||||
def broadcast_update(active_json: dict, closed_json: Optional[dict]):
|
||||
"""
|
||||
Callback executed by MT5Collector when a new tick is processed.
|
||||
Schedules WebSocket sends on the event loop.
|
||||
"""
|
||||
message = {
|
||||
"type": "tick",
|
||||
"active": active_json,
|
||||
"closed": closed_json
|
||||
}
|
||||
for connection in list(active_connections):
|
||||
try:
|
||||
asyncio.create_task(connection.send_json(message))
|
||||
except Exception as e:
|
||||
logger.error(f"Error sending update to client: {e}")
|
||||
|
||||
# Initialize MT5 Collector
|
||||
collector = MT5Collector(aggregator, on_update_callback=broadcast_update)
|
||||
|
||||
async def _start_collector_delayed():
|
||||
"""Wait a moment for the server to fully start, then begin polling MT5."""
|
||||
await asyncio.sleep(1)
|
||||
logger.info("Starting MT5 collector background task...")
|
||||
await collector.start()
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
global collector_task
|
||||
# Fire-and-forget: start collector AFTER yielding so uvicorn is ready
|
||||
collector_task = asyncio.create_task(_start_collector_delayed())
|
||||
logger.info("Server is starting up...")
|
||||
yield
|
||||
# Shutdown
|
||||
logger.info("Stopping MT5 collector task...")
|
||||
collector.running = False
|
||||
collector_task.cancel()
|
||||
try:
|
||||
await collector_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
await collector.disconnect_mt5()
|
||||
|
||||
app = FastAPI(title="YuClusters Local Server", lifespan=lifespan)
|
||||
|
||||
# Add CORS Middleware so local frontend can query history
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
@app.get("/history")
|
||||
async def get_history():
|
||||
"""Returns the buffer of historical closed clusters."""
|
||||
return aggregator.history
|
||||
|
||||
@app.websocket("/ws")
|
||||
async def websocket_endpoint(websocket: WebSocket):
|
||||
await websocket.accept()
|
||||
active_connections.add(websocket)
|
||||
logger.info(f"Client connected. Active connections: {len(active_connections)}")
|
||||
|
||||
# Send the current active cluster state on connection
|
||||
try:
|
||||
await websocket.send_json({
|
||||
"type": "init",
|
||||
"active": aggregator.active_cluster.to_json()
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"Error sending init state: {e}")
|
||||
|
||||
try:
|
||||
while True:
|
||||
await websocket.receive_text()
|
||||
except WebSocketDisconnect:
|
||||
active_connections.discard(websocket)
|
||||
logger.info(f"Client disconnected. Active connections: {len(active_connections)}")
|
||||
except Exception as e:
|
||||
logger.error(f"WebSocket error: {e}")
|
||||
active_connections.discard(websocket)
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
logger.info(f"Starting YuClusters server on port {settings.WS_PORT}...")
|
||||
uvicorn.run(
|
||||
"backend.server:app",
|
||||
host="0.0.0.0",
|
||||
port=settings.WS_PORT,
|
||||
log_level="info",
|
||||
)
|
||||
Reference in New Issue
Block a user