139 lines
5.2 KiB
Python
139 lines
5.2 KiB
Python
import mplfinance as mpf
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg, NavigationToolbar2Tk
|
|
import matplotlib.animation as animation
|
|
import ccxt
|
|
from matplotlib.dates import date2num
|
|
import threading
|
|
import tkinter as tk
|
|
from tkinter import ttk
|
|
|
|
class ChartView:
|
|
def __init__(self, exchange_ids, symbol='BTC/USDT', timeframe='1m', limit=100):
|
|
self.exchange_ids = exchange_ids
|
|
self.symbol = symbol
|
|
self.timeframe = timeframe
|
|
self.limit = limit
|
|
self.running = False
|
|
self.fig = None
|
|
self.ax1 = None
|
|
self.ax2 = None
|
|
self.animation = None
|
|
self.current_exchange_id = exchange_ids[0]
|
|
self.root = None
|
|
self.canvas = None
|
|
|
|
# Create exchanges dictionary to avoid reconnecting each time
|
|
self.exchanges = {}
|
|
for exchange_id in exchange_ids:
|
|
self.connect_to_exchange(exchange_id)
|
|
|
|
def connect_to_exchange(self, exchange_id):
|
|
if exchange_id in self.exchanges:
|
|
self.exchange = self.exchanges[exchange_id]
|
|
return
|
|
|
|
try:
|
|
exchange_class = getattr(ccxt, exchange_id)
|
|
exchange = exchange_class({
|
|
'enableRateLimit': True,
|
|
})
|
|
exchange.load_markets()
|
|
|
|
if self.symbol not in exchange.markets:
|
|
raise Exception(f"Symbol {self.symbol} not found on {exchange_id}")
|
|
|
|
self.exchanges[exchange_id] = exchange
|
|
self.exchange = exchange
|
|
except Exception as e:
|
|
raise Exception(f"Failed to connect to {exchange_id}: {str(e)}")
|
|
|
|
def fetch_ohlcv(self, exchange_id):
|
|
try:
|
|
exchange = self.exchanges[exchange_id]
|
|
ohlcv = exchange.fetch_ohlcv(self.symbol, self.timeframe, limit=self.limit)
|
|
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
|
|
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
|
|
df.set_index('timestamp', inplace=True)
|
|
return df
|
|
except Exception as e:
|
|
print(f"Error fetching OHLCV data from {exchange_id}: {str(e)}")
|
|
return None
|
|
|
|
def update_chart(self):
|
|
if not self.running:
|
|
return
|
|
|
|
df = self.fetch_ohlcv(self.current_exchange_id)
|
|
if df is not None:
|
|
self.ax1.clear()
|
|
self.ax2.clear()
|
|
|
|
mpf.plot(df, type='candle', ax=self.ax1, volume=self.ax2,
|
|
style='yahoo', xrotation=0, ylabel='Price',
|
|
ylabel_lower='Volume', show_nontrading=False)
|
|
|
|
latest_price = df['close'].iloc[-1]
|
|
self.ax1.set_title(f'{self.symbol} on {self.current_exchange_id} - Last: ${latest_price:.2f}')
|
|
|
|
# Refresh the canvas
|
|
self.canvas.draw()
|
|
|
|
def on_exchange_change(self, event=None):
|
|
selected_exchange = self.exchange_var.get()
|
|
if selected_exchange != self.current_exchange_id:
|
|
self.current_exchange_id = selected_exchange
|
|
self.update_chart()
|
|
|
|
def start_gui(self):
|
|
self.root = tk.Tk()
|
|
self.root.title("Crypto Chart Viewer")
|
|
self.root.geometry("1200x800")
|
|
|
|
# Create frame for controls
|
|
control_frame = ttk.Frame(self.root)
|
|
control_frame.pack(side=tk.TOP, fill=tk.X, padx=10, pady=5)
|
|
|
|
# Dropdown for selecting exchange
|
|
ttk.Label(control_frame, text="Exchange:").pack(side=tk.LEFT, padx=(0, 5))
|
|
self.exchange_var = tk.StringVar(value=self.current_exchange_id)
|
|
exchange_dropdown = ttk.Combobox(control_frame, textvariable=self.exchange_var, values=self.exchange_ids, width=15)
|
|
exchange_dropdown.pack(side=tk.LEFT, padx=(0, 10))
|
|
exchange_dropdown.bind("<<ComboboxSelected>>", self.on_exchange_change)
|
|
|
|
# Create the figure
|
|
self.fig = plt.Figure(figsize=(12, 8), dpi=100)
|
|
self.ax1 = self.fig.add_subplot(6, 1, (1, 4))
|
|
self.ax2 = self.fig.add_subplot(6, 1, (5, 6), sharex=self.ax1)
|
|
|
|
# Create the canvas to display the figure
|
|
self.canvas = FigureCanvasTkAgg(self.fig, master=self.root)
|
|
self.canvas.get_tk_widget().pack(side=tk.TOP, fill=tk.BOTH, expand=True)
|
|
|
|
# Add toolbar
|
|
toolbar = NavigationToolbar2Tk(self.canvas, self.root)
|
|
toolbar.update()
|
|
self.canvas.get_tk_widget().pack(side=tk.TOP, fill=tk.BOTH, expand=True)
|
|
|
|
# Initial chart update
|
|
self.running = True
|
|
self.update_chart()
|
|
|
|
# Set up periodic updates
|
|
def update():
|
|
if self.running:
|
|
self.update_chart()
|
|
self.root.after(10000, update) # Update every 10 seconds
|
|
|
|
self.root.after(10000, update)
|
|
|
|
# Handle window close
|
|
def on_closing():
|
|
self.running = False
|
|
self.root.destroy()
|
|
|
|
self.root.protocol("WM_DELETE_WINDOW", on_closing)
|
|
|
|
# Start the Tkinter event loop
|
|
self.root.mainloop() |