Umfangreiches Trading-Skript in Python

Erfahre, wie du automatisierten Handel mit der Fxview API implementierst

Highlights

API-Authentifizierung und Datenabruf: Sichere Verbindung zum Fxview API-Server inklusive Token-Generierung und Echtzeit-Datenabruf.
Technische Indikatoren und Handelslogik: Implementierung von Strategien wie Moving Average Crossover, RSI und Bollinger Bands sowie die Signalgenerierung.
Order-Management und Risikokontrolle: Platzierung von Marktorders, Stop-Loss/Take-Profit-Berechnungen und Echtzeit Order-Updates über WebSockets.

Einführung

Dieses umfassende Trading-Skript in Python dient als Grundlage für die Entwicklung eigener automatisierter Handelsstrategien unter Nutzung der Fxview API. Das Skript demonstriert umfangreiche Funktionalitäten von der API-Authentifizierung über den Abruf historischer und Echtzeit-Marktdaten bis hin zu einer vollständigen Handelslogik einschließlich technischer Indikatoren, Risikomanagement und Orderausführung.

Beachten Sie, dass der Handel mit Forex, CFDs und anderen Finanzinstrumenten mit hohen Risiken verbunden ist. Es wird dringend empfohlen, das Skript zunächst in einer Demo-Umgebung zu testen, bevor Sie es im Live-Handel einsetzen. Außerdem sollten Sie Ihre API-Zugangsdaten sicher verwahren und alle Parameter (wie z.B. Risikoprozent, Zeitrahmen, und Strategieparameter) an Ihre spezifischen Anforderungen anpassen.

Skript-Übersicht

Das folgende Python-Skript umfasst die folgenden Kernkomponenten:

1. API-Authentifizierung und Verbindung

Eine sichere Verbindung zum Fxview API-Server wird hergestellt, indem ein Authentifizierungstoken angefordert wird. Das Skript verwendet hierfür entweder die REST-API oder einen Digest-Authentication-Mechanismus. Eine erfolgreiche Authentifizierung ist Voraussetzung für alle weiteren Schritte, wie den Abruf von Marktdaten und das Platzieren von Orders.

2. Abruf von Marktdaten und Echtzeit-Updates

Historische Marktdaten werden über die REST-API abgerufen, um technische Indikatoren zu berechnen. Zusätzlich wird eine WebSocket-Verbindung eingerichtet, um in Echtzeit über Kursänderungen und Order-Updates informiert zu werden.

3. Berechnung technischer Indikatoren

Das Skript umfasst die Berechnung von Indikatoren, wie gleitende Durchschnitte (für Moving Average Crossover), den RSI (Relative Strength Index) und Bollinger Bänder. Diese Indikatoren helfen bei der Ableitung von Handelssignalen.

4. Handelslogik und Order-Management

Basierend auf den berechneten Indikatoren wird eine Handelslogik implementiert. Das Skript überprüft, ob Signale zum Kauf oder Verkauf vorliegen und führt diese gegebenenfalls automatisch durch. Dabei werden Stop-Loss und Take-Profit dynamisch berechnet, um das Risiko zu kontrollieren.

5. Risikomanagement

Eine wesentliche Komponente des Skripts ist die Berechnung der Positionsgröße. Dies geschieht anhand des Kontostandes unter Berücksichtigung eines festgelegten Risikoprozentsatzes, wodurch das Verlustrisiko bei einzelnen Trades minimiert wird.

Das Trading-Skript im Detail

Vorbereitungen und Bibliotheken

Bevor Sie das Skript ausführen, stellen Sie bitte sicher, dass Sie Python sowie alle notwendigen Bibliotheken installiert haben. Es wird empfohlen, ein virtuelles Environment zu verwenden. Die wichtigsten Pakete sind:

requests
json
pandas
numpy
matplotlib
websocket-client
hashlib
logging
threading

Installieren Sie diese mit pip:


# Installation der benötigten Pakete
pip install requests pandas numpy matplotlib websocket-client

Das Skript

Im Folgenden wird das Trading-Skript in Python vorgestellt. Alle API-Endpunkte, Parameter und Authentifizierungsmethoden wurden entsprechend der Fxview API-Dokumentation integriert. Passen Sie bitte die Platzhalter wie SERVER_ADDRESS, API_KEY und API_SECRET entsprechend Ihrer Zugangsdaten an.


# Importierung der notwendigen Bibliotheken
import requests
import json
import time
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import hashlib
import websocket
import threading
import logging
from datetime import datetime

# Logging-Konfiguration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("trading_bot.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("FxviewTradingBot")

class FxviewTradingBot:
    def __init__(self, server_address, api_key, api_secret, symbol="EURUSD", timeframe="1H", risk_percent=2.0):
        """
        Initialisiert den Trading Bot mit den notwendigen Parametern.
        :param server_address: Adresse des API-Servers
        :param api_key: API-Schlüssel
        :param api_secret: API-Geheimnis
        :param symbol: Handelssymbol (z.B. EURUSD)
        :param timeframe: Zeitrahmen (z.B. 1H, 4H, 1D)
        :param risk_percent: Prozentsatz des Kontos, der pro Trade riskiert wird
        """
        self.base_url = f"http://{server_address}:10001/api/v2"
        self.ws_url = f"ws://{server_address}:10006"
        self.api_key = api_key
        self.api_secret = api_secret
        self.symbol = symbol
        self.timeframe = timeframe
        self.risk_percent = risk_percent
        self.token = None
        self.account_info = None
        self.market_data = pd.DataFrame()
        self.open_trades = []
        self.ws = None
        self.ws_thread = None
        self.running = False
        # Strategie-Parameter
        self.strategy_params = {
            "ma_crossover": {"fast_ma": 9, "slow_ma": 21},
            "rsi": {"period": 14, "overbought": 70, "oversold": 30},
            "bollinger": {"period": 20, "std_dev": 2}
        }
        logger.info(f"Bot initialisiert für Symbol: {symbol}")
    
    def get_authentication_token(self):
        """Holt ein Authentifizierungstoken vom API-Server"""
        try:
            url = f"{self.base_url}/auth/token?lifetime=60"
            response = requests.get(url)
            if response.status_code == 200:
                self.token = response.json().get("result")
                logger.info("Token erfolgreich erhalten")
                return True
            # Digest Authentifizierung, wenn erforderlich
            if response.status_code == 401:
                auth_header = response.headers.get("www-authenticate")
                if auth_header and "Digest" in auth_header:
                    realm = auth_header.split('realm="')[1].split('"')[0]
                    nonce = auth_header.split('nonce="')[1].split('"')[0]
                    ha1 = hashlib.md5(f"{self.api_key}:{realm}:{self.api_secret}".encode()).hexdigest()
                    ha2 = hashlib.md5("GET:/api/v2/auth/token".encode()).hexdigest()
                    response_hash = hashlib.md5(f"{ha1}:{nonce}:{ha2}".encode()).hexdigest()
                    headers = {
                        'Authorization': f'Digest username="{self.api_key}", realm="{realm}", nonce="{nonce}", uri="/api/v2/auth/token", response="{response_hash}"'
                    }
                    response = requests.get(url, headers=headers)
                    if response.status_code == 200:
                        self.token = response.json().get("result")
                        logger.info("Token nach Digest Auth erfolgreich erhalten")
                        return True
            logger.error(f"Fehler beim Abrufen des Tokens: {response.status_code}")
            return False
        except Exception as e:
            logger.error(f"Ausnahme beim Holen des Tokens: {str(e)}")
            return False
    
    def get_account_info(self):
        """Holt Kontoinformationen vom API-Server"""
        if not self.token:
            logger.error("Kein Token vorhanden")
            return False
        try:
            headers = {"Authorization": f"Bearer {self.token}"}
            response = requests.get(f"{self.base_url}/account", headers=headers)
            if response.status_code == 200:
                data = response.json()
                if data.get("success"):
                    self.account_info = data.get("result")
                    logger.info(f"Kontoinformationen erhalten: Balance = {self.account_info.get('balance')}")
                    return True
            logger.error("Fehler beim Abrufen der Kontoinformationen")
            return False
        except Exception as e:
            logger.error(f"Ausnahme beim Abrufen der Kontoinformationen: {str(e)}")
            return False
    
    def get_market_data(self, limit=100):
        """Holt historische Marktdaten, um technische Indikatoren zu berechnen"""
        if not self.token:
            logger.error("Kein Token vorhanden")
            return False
        try:
            headers = {"Authorization": f"Bearer {self.token}"}
            params = {"symbol": self.symbol, "timeframe": self.timeframe, "limit": limit}
            response = requests.get(f"{self.base_url}/market/history", headers=headers, params=params)
            if response.status_code == 200:
                data = response.json()
                if data.get("success"):
                    df = pd.DataFrame(data.get("result"))
                    df["time"] = pd.to_datetime(df["time"], unit="ms")
                    df = df.set_index("time").sort_index()
                    self.market_data = df
                    logger.info(f"Historische Marktdaten empfangen: {len(df)} Kerzen")
                    return True
            logger.error("Fehler beim Abrufen der Marktdaten")
            return False
        except Exception as e:
            logger.error(f"Ausnahme beim Abrufen der Marktdaten: {str(e)}")
            return False
    
    def calculate_indicators(self, strategy="ma_crossover"):
        """Berechnet anhand der Marktdaten technische Indikatoren"""
        if self.market_data.empty:
            logger.error("Keine Marktdaten vorhanden")
            return False
        try:
            df = self.market_data.copy()
            if strategy == "ma_crossover":
                fast_ma = self.strategy_params["ma_crossover"]["fast_ma"]
                slow_ma = self.strategy_params["ma_crossover"]["slow_ma"]
                df[f"MA{fast_ma}"] = df["close"].rolling(window=fast_ma).mean()
                df[f"MA{slow_ma}"] = df["close"].rolling(window=slow_ma).mean()
                df["signal"] = 0
                df.loc[df[f"MA{fast_ma}"] > df[f"MA{slow_ma}"], "signal"] = 1
                df.loc[df[f"MA{fast_ma}"] < df[f"MA{slow_ma}"], "signal"] = -1
                df["signal_change"] = df["signal"].diff()
            elif strategy == "rsi":
                period = self.strategy_params["rsi"]["period"]
                delta = df["close"].diff()
                gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
                loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
                rs = gain / loss
                df["RSI"] = 100 - (100 / (1 + rs))
                df["signal"] = 0
                df.loc[df["RSI"] < self.strategy_params["rsi"]["oversold"], "signal"] = 1
                df.loc[df["RSI"] > self.strategy_params["rsi"]["overbought"], "signal"] = -1
                df["signal_change"] = df["signal"].diff()
            elif strategy == "bollinger":
                period = self.strategy_params["bollinger"]["period"]
                std_dev = self.strategy_params["bollinger"]["std_dev"]
                df["MA"] = df["close"].rolling(window=period).mean()
                df["STD"] = df["close"].rolling(window=period).std()
                df["upper_band"] = df["MA"] + (std_dev * df["STD"])
                df["lower_band"] = df["MA"] - (std_dev * df["STD"])
                df["signal"] = 0
                df.loc[df["close"] < df["lower_band"], "signal"] = 1
                df.loc[df["close"] > df["upper_band"], "signal"] = -1
                df["signal_change"] = df["signal"].diff()
            self.market_data = df
            logger.info(f"Technische Indikatoren für {strategy} berechnet")
            return True
        except Exception as e:
            logger.error(f"Fehler bei der Berechnung technischer Indikatoren: {str(e)}")
            return False
    
    def generate_trading_signal(self):
        """Erstellt ein Handelssignal basierend auf den Indikatoren"""
        if self.market_data.empty:
            logger.error("Keine Marktdaten zum Generieren des Signals")
            return None
        try:
            latest = self.market_data.iloc[-1]
            if "signal_change" in latest and latest["signal_change"] != 0:
                if latest["signal_change"] > 0:
                    return {"action": "buy", "price": latest["close"]}
                elif latest["signal_change"] < 0:
                    return {"action": "sell", "price": latest["close"]}
            return None
        except Exception as e:
            logger.error(f"Fehler beim Generieren des Handelssignals: {str(e)}")
            return None
    
    def calculate_position_size(self, price, stop_loss_pips):
        """Berechnet die Positionsgröße basierend auf dem Kontostand und Risiko"""
        try:
            account_balance = float(self.account_info.get("balance", 0))
            risk_amount = account_balance * (self.risk_percent / 100)
            pip_value = 0.0001 if "JPY" not in self.symbol else 0.01
            stop_loss_amount = stop_loss_pips * pip_value
            pos_size = risk_amount / stop_loss_amount
            # Runden auf 0.01-Schritte
            pos_size = round(pos_size / 0.01) * 0.01
            min_lot, max_lot = 0.01, 100.0
            pos_size = max(min(pos_size, max_lot), min_lot)
            logger.info(f"Positionsgröße berechnet: {pos_size} Lots")
            return pos_size
        except Exception as e:
            logger.error(f"Fehler bei der Berechnung der Positionsgröße: {str(e)}")
            return 0.01
    
    def place_market_order(self, action, lot_size, stop_loss=None, take_profit=None):
        """Platziert eine Marktorder anhand der Order-Parameter"""
        if not self.token:
            logger.error("Kein Token vorhanden, Order kann nicht platziert werden")
            return False
        try:
            headers = {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}
            order_type = "buy" if action == "buy" else "sell"
            data = {
                "symbol": self.symbol,
                "type": order_type,
                "volume": lot_size
            }
            if stop_loss:
                data["sl"] = stop_loss
            if take_profit:
                data["tp"] = take_profit
            response = requests.post(f"{self.base_url}/trading/placemarket", headers=headers, json=data)
            if response.status_code == 200:
                order_data = response.json()
                if order_data.get("success"):
                    logger.info(f"{action.upper()}-Order erfolgreich platziert: {order_data.get('result')}")
                    return order_data.get("result").get("order")
            logger.error(f"Fehler beim Platzieren der Order: {response.text}")
            return False
        except Exception as e:
            logger.error(f"Ausnahme bei Orderplatzierung: {str(e)}")
            return False
    
    def close_trade(self, trade_id):
        """Schließt einen offenen Trade anhand seiner ID"""
        if not self.token:
            logger.error("Kein Token vorhanden, Trade kann nicht geschlossen werden")
            return False
        try:
            headers = {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}
            data = {"id": trade_id}
            response = requests.post(f"{self.base_url}/trading/closetrade", headers=headers, json=data)
            if response.status_code == 200:
                result = response.json()
                if result.get("success"):
                    logger.info(f"Trade {trade_id} geschlossen")
                    return True
            logger.error(f"Fehler beim Schließen des Trades {trade_id}: {response.text}")
            return False
        except Exception as e:
            logger.error(f"Ausnahme beim Schließen des Trades: {str(e)}")
            return False
    
    def init_websocket(self):
        """Initialisiert WebSocket-Verbindung für Echtzeit-Daten"""
        if not self.token:
            logger.error("Kein Token vorhanden, WebSocket kann nicht initialisiert werden")
            return False
        try:
            def on_message(ws, message):
                data = json.loads(message)
                if "tick" in data:
                    tick = data["tick"]
                    logger.debug(f"WebSocket Tick: {tick.get('symbol')} Bid: {tick.get('bid')} Ask: {tick.get('ask')}")
                elif "trade" in data:
                    trade = data["trade"]
                    logger.info(f"Trade Update: ID {trade.get('id')} Status: {trade.get('status')}")
            def on_error(ws, error):
                logger.error(f"WebSocket Fehler: {str(error)}")
            def on_close(ws, close_status_code, close_msg):
                logger.info(f"WebSocket geschlossen: {close_msg}")
            def on_open(ws):
                logger.info("WebSocket verbunden")
                subscription = {
                    "action": "subscribe",
                    "topics": [{"name": "ticks", "symbols": [self.symbol]}, {"name": "trades"}],
                    "token": self.token
                }
                ws.send(json.dumps(subscription))
            self.ws = websocket.WebSocketApp(
                self.ws_url, on_message=on_message, on_error=on_error,
                on_close=on_close, on_open=on_open
            )
            self.ws_thread = threading.Thread(target=self.ws.run_forever)
            self.ws_thread.daemon = True
            self.ws_thread.start()
            logger.info("WebSocket-Thread gestartet")
            return True
        except Exception as e:
            logger.error(f"Fehler beim Initialisieren des WebSockets: {str(e)}")
            return False
    
    def run(self):
        """Hauptschleife des Trading-Bots zur kontinuierlichen Ausführung"""
        logger.info("Trading-Bot startet")
        if not self.get_authentication_token():
            logger.error("Token konnte nicht abgerufen werden. Bot beendet sich.")
            return
        if not self.get_account_info():
            logger.error("Kontoinformationen konnten nicht abgerufen werden. Bot beendet sich.")
            return
        if not self.init_websocket():
            logger.warning("WebSocket konnte nicht initialisiert werden, Handel läuft weiter ohne Echtzeit-Updates.")
        self.running = True
        while self.running:
            try:
                if not self.get_market_data():
                    logger.error("Fehler beim Abruf der Marktdaten. Nächster Zyklus in 60 Sekunden.")
                    time.sleep(60)
                    continue
                if not self.calculate_indicators(strategy="ma_crossover"):
                    logger.error("Fehler bei der Berechnung der Indikatoren. Nächster Zyklus in 60 Sekunden.")
                    time.sleep(60)
                    continue
                signal = self.generate_trading_signal()
                if signal:
                    logger.info(f"Handelssignal generiert: {signal.get('action')} bei {signal.get('price')}")
                    current_trades = self.open_trades  # Annahme: vorherige Abruf in get_open_trades implementieren
                    existing = False
                    for trade in current_trades:
                        if trade.get("symbol") == self.symbol:
                            existing = True
                            if (signal.get("action") == "buy" and trade.get("type") == "sell") or \
                               (signal.get("action") == "sell" and trade.get("type") == "buy"):
                                logger.info(f"Schließe bestehende Position: {trade.get('id')}")
                                self.close_trade(trade.get("id"))
                    if not existing:
                        stop_loss_pips = 50  # Beispielwert, anpassen!
                        lot_size = self.calculate_position_size(signal.get("price"), stop_loss_pips)
                        pip_value = 0.0001 if "JPY" not in self.symbol else 0.01
                        if signal.get("action") == "buy":
                            stop_loss_price = signal.get("price") - (stop_loss_pips * pip_value)
                            take_profit_price = signal.get("price") + (2 * stop_loss_pips * pip_value)
                        else:
                            stop_loss_price = signal.get("price") + (stop_loss_pips * pip_value)
                            take_profit_price = signal.get("price") - (2 * stop_loss_pips * pip_value)
                        order_id = self.place_market_order(
                            signal.get("action"),
                            lot_size,
                            stop_loss=round(stop_loss_price, 5),
                            take_profit=round(take_profit_price, 5)
                        )
                        if order_id:
                            logger.info(f"{signal.get('action').upper()}-Order platziert, Order-ID: {order_id}")
                wait_time = 900  # 15 Minuten Zyklus
                logger.info(f"Nächster Zyklus in {wait_time} Sekunden")
                time.sleep(wait_time)
            except Exception as e:
                logger.error(f"Unerwarteter Fehler im Hauptzyklus: {str(e)}")
                time.sleep(60)
    
    def stop(self):
        """Stoppt den Trading-Bot"""
        logger.info("Trading-Bot wird gestoppt")
        self.running = False
        if self.ws:
            self.ws.close()
        if self.ws_thread and self.ws_thread.is_alive():
            self.ws_thread.join(timeout=5)
        logger.info("Trading-Bot wurde gestoppt")

if __name__ == "__main__":
    # Konfiguration der API und der Handelsparameter
    SERVER_ADDRESS = "s245.sysfx.com"  # Ersetzen Sie durch die tatsächliche Server-Adresse
    API_KEY = "YOUR_API_KEY"
    API_SECRET = "YOUR_API_SECRET"
    bot = FxviewTradingBot(
        server_address=SERVER_ADDRESS,
        api_key=API_KEY,
        api_secret=API_SECRET,
        symbol="EURUSD",
        timeframe="1H",
        risk_percent=2.0
    )
    try:
        bot.run()
    except KeyboardInterrupt:
        bot.stop()
        print("Bot wurde manuell gestoppt")
    except Exception as e:
        print(f"Ein unerwarteter Fehler ist aufgetreten: {str(e)}")
        bot.stop()

Erklärung und Erweiterungsmöglichkeiten

Das obige Skript legt den Grundstein für ein automatisiertes Handelssystem mit der Fxview API. Nachfolgend einige Erklärungen zu den einzelnen Modulen sowie Erweiterungsmöglichkeiten:

API-Authentifizierung

Die Methode zur Token-Generierung stellt die Grundlage für alle weiteren API-Aufrufe dar. Durch den Einsatz von Digest Authentication kann das Skript auch in Umgebungen verwendet werden, welche erhöhte Sicherheitsanforderungen stellen.

Datenabruf und Indikatoren

Historische Marktdaten werden verwendet, um technische Indikatoren zu berechnen. Dabei können Sie die Wahl zwischen verschiedenen Strategien treffen:

Strategie	Indikator(en)	Anwendung
MA Crossover	Gleitender Durchschnitt (Fast und Slow)	Erzeugt Kauf-/Verkaufssignale bei Überschreitungen
RSI	Relative Strength Index	Identifiziert überkaufte oder überverkaufte Zustände
Bollinger Bands	Gleitender Durchschnitt und Standardabweichung	Liefert Signale basierend auf Bandausbrüchen

Diese Tabelle veranschaulicht, wie die verschiedenen Strategien jeweils zu Handelssignalen führen können.

Handelslogik und Risikomanagement

Die Handelslogik in der Hauptschleife des Bots beinhaltet das Abrufen aktueller Marktdaten, die Berechnung der Indikatoren und die Generierung eines Handelssignals. Wird ein Signal erzeugt, so vergleicht der Bot, ob bereits eine offene Position zu diesem Symbol existiert. Entsprechend des Risikomanagements wird die optimale Lot-Größe errechnet und Orderparameter wie Stop-Loss und Take-Profit dynamisch festgelegt. Diese Parameter sind essenziell, um das Verlustrisiko zu minimieren und ein ausgewogenes Chance-Risiko-Verhältnis zu gewährleisten.

Echtzeit-Updates via WebSocket

Durch die WebSocket-Integration erhält der Bot in Echtzeit Updates zu Kursbewegungen und Order-Änderungen. Dies verbessert die Reaktionsfähigkeit des Handelssystems erheblich, da Anpassungen in kürzeren Zykluszeiten realisiert werden können. Sollten Sie die Performance verbessern wollen, können Sie das Abonnement der WebSocket-Themen noch weiter optimieren.

Abschluss und Schlussfolgerungen

Dieses Trading-Skript bietet eine umfassende Grundlage zur Entwicklung und Implementierung automatisierter Trading-Strategien unter der Verwendung der Fxview API. Es integriert essentielle Funktionen wie API-Authentifizierung, Datenabruf, technische Indikatoren, Signalverarbeitung, Orderausführung und Risikomanagement. Die modular aufgebaute Struktur ermöglicht es Ihnen, einzelne Komponenten – etwa die Strategie oder das Risikomanagement – flexibel zu erweitern oder auszutauschen.

Vor dem Live-Einsatz sollte das Skript in verschiedenen Marktbedingungen und mit historischen Daten eingehend getestet werden. Zusätzlich können weitere Sicherheits- und Optimierungsmaßnahmen implementiert werden, um die Stabilität und Sicherheit des Handelssystems zu erhöhen.

Zusammenfassend lässt sich sagen, dass dieses Skript als leistungsfähige Basis dient, um automatisierten Handel effizient und sicher zu realisieren. Nutzen Sie die Chance, die verschiedenen Module zu erweitern und exakt auf Ihre individuellen Anforderungen anzupassen.

Fazit

Das vorgestellte Trading-Skript demonstriert einen vollständigen Ablauf von der API-Authentifizierung über den Abruf historischer und Echtzeit-Daten bis zur Ausführung von Handelsstrategien mittels technischer Indikatoren. Dabei spielt das Risikomanagement eine zentrale Rolle, um Verluste zu begrenzen und den Gesamtansatz zu optimieren. Benutzer können dieses Skript an ihre individuellen Anforderungen anpassen, indem sie Parameter, Strategien und Sicherheitsfunktionen erweitern.