Enseñarle a actualizar el colector de mercado backtest la fuente de datos personalizados

Artículo anteriorEnseñarle a implementar un coleccionista de cotizaciones de mercadoHemos implementado un programa robótico que recopila cotizaciones del mercado juntos.

¿Cómo usar los datos del mercado después de que los recopilamos? se utilizará para el sistema de backtest. Confiando en la función de fuente de datos personalizada del sistema de backtest de la plataforma FMZ, podemos usar directamente los datos recopilados como la fuente de datos del sistema de backtest, para que podamos dejar que el sistema de backtest use en cualquier mercado donde queramos backtestar datos históricos.

Por lo tanto, podemos dar el Market Quote Collector una actualización! dejar que el recolector de mercado también puede servir como una fuente de datos personalizados para proporcionar datos al sistema de backtest.

Prepárate

Es diferente del trabajo de preparación en el artículo anterior. La última vez fue un programa docker que se ejecuta en mi computadora MAC local, instalando la base de datos mongodb para iniciar el servicio de base de datos. Esta vez cambiamos el entorno operativo a VPS y usamos el servidor Alibaba Cloud Linux para ejecutar nuestro conjunto de programas.

• Base de datos de Mongodb

Como en el artículo anterior, necesitamos instalar la base de datos mongodb en el dispositivo donde se está ejecutando el programa market collector y iniciar el servicio. Es básicamente lo mismo que instalar mongodb en un ordenador MAC. Hay muchos tutoriales en Internet, puedes buscarlo en Google, es muy simple.

• Instalar Python 3

El programa utiliza python3, tenga cuidado con el uso de algunas bibliotecas de python, si no están instaladas, primero debe instalarlas.

Pymongo - ¿ Qué es esto? - ¿ Qué pasa?

• El muelle

Un FMZ en funcionamiento será suficiente.

Transforma el Colector de cotizaciones de mercado

El coleccionista de cotizaciones de mercado es este:https://www.fmz.com/strategy/199120(RecordsCollector) estrategia.

Vamos a hacer algunas modificaciones:

Antes de que el programa ingrese al bucle while para recopilar datos, se utiliza una biblioteca de múltiples hilos, y la ejecución simultánea inicia un servicio para monitorear la solicitud de datos del sistema de backtest de la plataforma FMZ. (Se pueden ignorar otros detalles)

RecordsCollector (actualización para proporcionar una función de fuente de datos personalizada)

import _thread
import pymongo
import json
import math
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse

def url2Dict(url):
    query = urlparse(url).query  
    params = parse_qs(query)  
    result = {key: params[key][0] for key in params}  
    return result

class Provider(BaseHTTPRequestHandler):
    def do_GET(self):
        try:
            self.send_response(200)
            self.send_header("Content-type", "application/json")
            self.end_headers()

            dictParam = url2Dict(self.path)
            Log("The custom data source service receives the request, self.path:", self.path, "query parameter:", dictParam)
            
            # At present, the backtesting system can only select the exchange name from the list. When adding a custom data source, set it to Binance, that is: Binance
            exName = exchange.GetName()                                     
            # Note that period is the bottom K-line period
            tabName = "%s_%s" % ("records", int(int(dictParam["period"]) / 1000))  
            priceRatio = math.pow(10, int(dictParam["round"]))
            amountRatio = math.pow(10, int(dictParam["vround"]))
            fromTS = int(dictParam["from"]) * int(1000)
            toTS = int(dictParam["to"]) * int(1000)
            
            
            # Connect to the database
            Log("Connect to the database service to obtain data, the database:", exName, "table:", tabName)
            myDBClient = pymongo.MongoClient("mongodb://localhost:27017")
            ex_DB = myDBClient[exName]
            exRecords = ex_DB[tabName]
            
            
            # Request data
            data = {
                "schema" : ["time", "open", "high", "low", "close", "vol"],
                "data" : []
            }
            
            # Construct query condition: greater than a certain value{'age': {'$gt': 20}} Less than a certain value{'age': {'$lt': 20}}
            dbQuery = {"$and":[{'Time': {'$gt': fromTS}}, {'Time': {'$lt': toTS}}]}
            Log("Query conditions:", dbQuery, "Number of inquiries:", exRecords.find(dbQuery).count(), "Total number of databases:", exRecords.find().count())
            
            for x in exRecords.find(dbQuery).sort("Time"):
                # Need to process data accuracy according to request parameters round and vround
                bar = [x["Time"], int(x["Open"] * priceRatio), int(x["High"] * priceRatio), int(x["Low"] * priceRatio), int(x["Close"] * priceRatio), int(x["Volume"] * amountRatio)]
                data["data"].append(bar)
            
            Log("data:", data, "Respond to backtest system requests.")
            # Write data reply
            self.wfile.write(json.dumps(data).encode())
        except BaseException as e:
            Log("Provider do_GET error, e:", e)


def createServer(host):
    try:
        server = HTTPServer(host, Provider)
        Log("Starting server, listen at: %s:%s" % host)
        server.serve_forever()
    except BaseException as e:
        Log("createServer error, e:", e)
        raise Exception("stop")

def main():
    LogReset(1)
    exName = exchange.GetName()
    period = exchange.GetPeriod()
    Log("collect", exName, "Exchange K-line data,", "K line cycle:", period, "second")
    
    # Connect to the database service, service address mongodb://127.0.0.1:27017 See the settings of mongodb installed on the server
    Log("Connect to the mongodb service of the hosting device, mongodb://localhost:27017")
    myDBClient = pymongo.MongoClient("mongodb://localhost:27017")   
    # Create a database
    ex_DB = myDBClient[exName]
    
    # Print the current database table
    collist = ex_DB.list_collection_names()
    Log("mongodb ", exName, " collist:", collist)
    
    # Check if the table is deleted
    arrDropNames = json.loads(dropNames)
    if isinstance(arrDropNames, list):
        for i in range(len(arrDropNames)):
            dropName = arrDropNames[i]
            if isinstance(dropName, str):
                if not dropName in collist:
                    continue
                tab = ex_DB[dropName]
                Log("dropName:", dropName, "delete:", dropName)
                ret = tab.drop()
                collist = ex_DB.list_collection_names()
                if dropName in collist:
                    Log(dropName, "failed to delete")
                else :
                    Log(dropName, "successfully deleted")
    
    # Start a thread to provide a custom data source service
    try:
        # _thread.start_new_thread(createServer, (("localhost", 9090), ))     # local computer test
        _thread.start_new_thread(createServer, (("0.0.0.0", 9090), ))         # Test on VPS server
        Log("Open the custom data source service thread", "#FF0000")
    except BaseException as e:
        Log("Failed to start the custom data source service!")
        Log("Error message:", e)
        raise Exception("stop")
    
    # Create the records table
    ex_DB_Records = ex_DB["%s_%d" % ("records", period)]
    Log("Start collecting", exName, "K-line data", "cycle:", period, "Open (create) the database table:", "%s_%d" % ("records", period), "#FF0000")
    preBarTime = 0
    index = 1
    while True:
        r = _C(exchange.GetRecords)
        if len(r) < 2:
            Sleep(1000)
            continue
        if preBarTime == 0:
            # Write all BAR data for the first time
            for i in range(len(r) - 1):
                bar = r[i]
                # Write line by line, you need to determine whether the data already exists in the current database table, based on timestamp detection, if there is the data, then skip, if not write in
                retQuery = ex_DB_Records.find({"Time": bar["Time"]})
                if retQuery.count() > 0:
                    continue
                
                # Write bar to the database table
                ex_DB_Records.insert_one({"High": bar["High"], "Low": bar["Low"], "Open": bar["Open"], "Close": bar["Close"], "Time": bar["Time"], "Volume": bar["Volume"]})                
                index += 1
            preBarTime = r[-1]["Time"]
        elif preBarTime != r[-1]["Time"]:
            bar = r[-2]
            # Check before writing data, whether the data already exists, based on time stamp detection
            retQuery = ex_DB_Records.find({"Time": bar["Time"]})
            if retQuery.count() > 0:
                continue
            
            ex_DB_Records.insert_one({"High": bar["High"], "Low": bar["Low"], "Open": bar["Open"], "Close": bar["Close"], "Time": bar["Time"], "Volume": bar["Volume"]})
            index += 1
            preBarTime = r[-1]["Time"]
        LogStatus(_D(), "preBarTime:", preBarTime, "_D(preBarTime):", _D(preBarTime/1000), "index:", index)
        # adding drawing display
        ext.PlotRecords(r, "%s_%d" % ("records", period))
        Sleep(10000)

Prueba

Configurar el robot

Dirigir el robot, ejecutar el coleccionista de cotizaciones de mercado.

Abra una estrategia de prueba para backtest. Por ejemplo:

function main() {
    Log(exchange.GetRecords())
    Log(exchange.GetRecords())
    Log(exchange.GetRecords())
    Log(exchange.GetRecords())
    Log(exchange.GetRecords())
    Log(exchange.GetRecords())
    Log(exchange.GetRecords().length)
}

Configurar la opción de backtest, configurar el intercambio a Binance porque la fuente de datos personalizados temporales aún no puede formular un nombre de intercambio por sí mismo, sólo se puede pedir prestado una de las configuraciones de intercambio en la lista, el backtest muestra que Binance, el real Es los datos del mercado de simulación de WexApp.

Compare si el gráfico generado por el sistema de backtest basado en el coleccionista de cotizaciones de mercado como fuente de datos personalizada es el mismo que el gráfico de línea K de 1 hora en la página de intercambio de wexApp.

De esta manera, el robot en el VPS puede recopilar datos de línea K por sí mismo, y podemos obtener los datos recopilados en cualquier momento y backtest directamente en el sistema de backtest.

Puede seguir expandiendo, por ejemplo, probar el nivel real de pruebas de retroceso de las fuentes de datos personalizadas, y la recolección de datos multi-variedad, multi-mercado y otras funciones.