my code :
from flask import Flask, render_template, request, redirect, url_for, session, flash
import ccxt
import os
import datetime as dt
import pandas as pd
import pandas_ta as ta
import okx.Account as Account
import okx
from websocket import WebSocketApp
app = Flask(name)
app.secret_key = os.urandom(24)
SUPPORTED_EXCHANGES = ['binance', 'coinbasepro', 'okx', 'kraken', 'kucoin', 'bitfinex']
Global cache for WebSocket connections
websocket_connections = {}
Function to create a WebSocket connection with OKX
def create_okx_websocket(api_key, api_secret, passphrase):
ws_url = "wss://ws.okx.com:8443/ws/v5/private"
ws = WebSocketApp(ws_url)
return ws
Function to create a connection with OKX
def create_okx_connection(api_key, api_secret, passphrase):
flag = "1" # Production trading: 0, demo trading: 1
return okx.Account.AccountAPI(api_key, api_secret, passphrase, False, flag, proxy=False)
Function to create a CCXT exchange instance
def create_exchange_instance(exchange_name, api_key, api_secret, passphrase=None):
exchange_params = {
'apiKey': api_key,
'secret': api_secret,
'timeout': 30000,
'enableRateLimit': True,
}
if passphrase:
exchange_params['password'] = passphrase
return getattr(ccxt, exchange_name)(exchange_params)
Function to get the connection status
def get_connection_status():
exchange_name = session.get("exchange_name")
if exchange_name:
return f"Connected to {exchange_name.capitalize()}"
else:
return "Not connected"
@app.route('/', methods=['GET', 'POST'])
def index():
error = None
if request.method == 'POST':
exchange_name = request.form['exchange_name']
api_key = request.form['api_key']
api_secret = request.form['api_secret']
passphrase = request.form['passphrase']
try:
if exchange_name == "okx":
connection = create_okx_connection(api_key, api_secret, passphrase)
websocket_connections[exchange_name] = create_okx_websocket(api_key, api_secret, passphrase)
session.update({
"exchange_name": exchange_name,
"api_key": api_key,
"api_secret": api_secret,
"passphrase": passphrase,
})
flash(f"Successfully connected to {exchange_name.capitalize()}!", "success")
return redirect(url_for('ia_control'))
else:
connection = create_exchange_instance(exchange_name, api_key, api_secret, passphrase)
session["exchange_connection"] = connection
flash(f"Successfully connected to {exchange_name.capitalize()}!", "success")
return redirect(url_for('ia_control'))
except Exception as e:
error = f"Error during connection: {e}"
return render_template(
'index.html',
supported_exchanges=SUPPORTED_EXCHANGES,
error=error,
get_connection_status=get_connection_status, # Pass the function, not the result
)
@app.route('/logout')
def logout():
session.clear() # Clear the session to log out
websocket_connections.clear() # Clear WebSocket connections
flash("You have been logged out.", "success")
return redirect(url_for('index'))
@app.route('/trade', methods=['GET', 'POST'])
def trade():
exchange_name = session.get('exchange_name')
api_key = session.get('api_key')
api_secret = session.get('api_secret')
passphrase = session.get('passphrase', None)
if not (exchange_name and api_key and api_secret):
flash("Please connect to an exchange to use trading.", "error")
return redirect(url_for('index'))
try:
exchange = create_exchange_instance(exchange_name, api_key, api_secret, passphrase)
except Exception as e:
flash(f"Error connecting to the exchange: {e}", "error")
return redirect(url_for('index'))
if request.method == 'POST':
symbol = request.form.get('symbol').upper()
amount = float(request.form.get('amount'))
stop_loss = request.form.get('stop_loss', None)
take_profit = request.form.get('take_profit', None)
try:
order = exchange.create_market_buy_order(symbol, amount)
if stop_loss:
# Logic for stop-loss
pass
if take_profit:
# Logic for take-profit
pass
flash("Market buy order placed successfully!", "success")
except Exception as e:
flash(f"Error placing order: {e}", "error")
return render_template('trade.html', get_connection_status=get_connection_status())
@app.route('/ia_control', methods=['GET'])
def ia_control():
exchange_name = session.get('exchange_name')
api_key = session.get('api_key')
api_secret = session.get('api_secret')
passphrase = session.get('passphrase', None)
if not (exchange_name and api_key and api_secret):
flash("Please connect to an exchange to access this page.", "error")
return redirect(url_for('index'))
try:
exchange = create_exchange_instance(exchange_name, api_key, api_secret, passphrase)
symbol = 'BTC/USDT'
timeframe = '1h'
since = dt.datetime.now() - dt.timedelta(days=7)
ohlcv = exchange.fetch_ohlcv(symbol, timeframe)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df['rsi'] = df.ta.rsi(length=14)
df['macd'], df['signal'], df['histogram'] = df.ta.macd(fast=12, slow=26, signal=9)
df['sma_50'] = df.ta.sma(length=50)
df['sma_200'] = df.ta.sma(length=200)
recommendations = []
if df['rsi'].iloc[-1] > 70:
recommendations.append("RSI indicates overbought.")
elif df['rsi'].iloc[-1] < 30:
recommendations.append("RSI indicates oversold.")
if df['macd'].iloc[-1] > df['signal']:
recommendations.append("MACD suggests bullish momentum.")
else:
recommendations.append("MACD indicates bearish momentum.")
return render_template(
'ia_control.html',
symbol=symbol,
dataframe=df,
recommendations=recommendations,
get_connection_status=get_connection_status(),
)
except Exception as e:
flash(f"Error connecting to the exchange: {e}", "error")
return redirect(url_for('index'))
if name == 'main':
app.run(debug=True)
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent