Perbincangan mengenai Kaedah Ujian Strategi Berdasarkan Random Ticker Generator

Pengantar

Sistem backtesting Platform Perdagangan Kuantum FMZ adalah sistem backtesting yang sentiasa berulang, dikemas kini dan dinaik taraf. Ia menambah fungsi dan mengoptimumkan prestasi secara beransur-ansur dari fungsi backtesting asas awal. Dengan perkembangan platform, sistem backtesting akan terus dioptimumkan dan dinaik taraf. Hari ini kita akan membincangkan topik berdasarkan sistem backtesting: Pengujian strategi berdasarkan ticker rawak

Permintaan

Dalam bidang perdagangan kuantitatif, pembangunan dan pengoptimuman strategi tidak dapat dipisahkan dari pengesahan data pasaran sebenar. Walau bagaimanapun, dalam aplikasi sebenar, disebabkan oleh persekitaran pasaran yang kompleks dan berubah, bergantung pada data sejarah untuk backtesting mungkin tidak mencukupi, seperti kekurangan liputan keadaan pasaran yang melampau atau senario khas. Oleh itu, merancang penjana pasaran rawak yang cekap telah menjadi alat yang berkesan untuk pemaju strategi kuantitatif.

Apabila kita perlu membiarkan strategi mengesan semula data sejarah pada bursa atau mata wang tertentu, kita boleh menggunakan sumber data rasmi platform FMZ untuk backtesting.

Kepentingan menggunakan data ticker rawak adalah:

• 1. Mengkaji kekuatan strategi Penjana ticker rawak boleh membuat pelbagai senario pasaran yang mungkin, termasuk turun naik yang melampau, turun naik yang rendah, pasaran trend, dan pasaran yang tidak menentu.

Adakah strategi boleh disesuaikan dengan perubahan trend dan turun naik? Adakah strategi akan mengalami kerugian besar dalam keadaan pasaran yang melampau?

• 2. Mengenal pasti kelemahan berpotensi dalam strategi Dengan mensimulasikan beberapa keadaan pasaran yang tidak normal (seperti peristiwa angsa hitam hipotetis), potensi kelemahan dalam strategi dapat ditemui dan ditingkatkan.

Adakah strategi terlalu bergantung pada struktur pasaran tertentu? Adakah terdapat risiko parameter yang terlalu sesuai?

• 3. Mengoptimumkan parameter strategi Data yang dihasilkan secara rawak menyediakan persekitaran ujian yang lebih pelbagai untuk pengoptimuman parameter strategi, tanpa perlu bergantung sepenuhnya pada data sejarah. Ini membolehkan anda mencari julat parameter strategi dengan lebih komprehensif dan mengelakkan terhad kepada corak pasaran tertentu dalam data sejarah.
• 4. Mengisi jurang dalam data sejarah Di sesetengah pasaran (seperti pasaran baru muncul atau pasaran perdagangan mata wang kecil), data sejarah mungkin tidak mencukupi untuk merangkumi semua keadaan pasaran yang mungkin.
• 5. Pembangunan iteratif yang pantas Menggunakan data rawak untuk ujian pantas boleh mempercepatkan pengulangan pembangunan strategi tanpa bergantung pada keadaan pasaran masa nyata atau pembersihan dan organisasi data yang memakan masa.

Walau bagaimanapun, ia juga perlu untuk menilai strategi secara rasional.

• 1. Walaupun penjana pasaran rawak berguna, kepentingan mereka bergantung kepada kualiti data yang dihasilkan dan reka bentuk senario sasaran:
• 2. Logik penjanaan perlu dekat dengan pasaran sebenar: Jika pasaran yang dihasilkan secara rawak benar-benar tidak berhubung dengan realiti, hasil ujian mungkin tidak mempunyai nilai rujukan. Sebagai contoh, penjana boleh direka berdasarkan ciri statistik pasaran sebenar (seperti pengedaran turun naik, nisbah trend).
• 3. Ia tidak boleh menggantikan sepenuhnya ujian data sebenar: data rawak hanya dapat melengkapkan pembangunan dan pengoptimuman strategi.

Setelah berkata begitu banyak, bagaimana kita boleh fabrikasi beberapa data? bagaimana kita boleh fabrikasi data untuk sistem backtesting untuk digunakan dengan mudah, cepat dan mudah?

Idea Reka Bentuk

Artikel ini direka untuk menyediakan titik permulaan untuk perbincangan dan menyediakan pengiraan penjanaan ticker rawak yang agak mudah. Sebenarnya, terdapat pelbagai algoritma simulasi, model data dan teknologi lain yang boleh digunakan. Oleh kerana ruang perbincangan terbatas, kami tidak akan menggunakan kaedah simulasi data yang kompleks.

Menggabungkan fungsi sumber data tersuai dari sistem backtesting platform, kami menulis program dalam Python.

• 1. Menghasilkan satu set data K-line secara rawak dan menulis mereka ke dalam fail CSV untuk rakaman berterusan, supaya data yang dihasilkan dapat disimpan.
• 2. Kemudian mencipta perkhidmatan untuk menyediakan sokongan sumber data untuk sistem backtesting.
• 3. Tampilkan data K-line yang dihasilkan dalam carta.

Untuk beberapa piawaian penjanaan dan penyimpanan fail data K-line, kawalan parameter berikut boleh ditakrifkan:

• Mod penjanaan data rawak Untuk simulasi jenis fluktuasi data K-line, reka bentuk mudah hanya dibuat menggunakan kebarangkalian nombor rawak positif dan negatif. Apabila data yang dihasilkan tidak banyak, ia mungkin tidak mencerminkan corak pasaran yang diperlukan. Jika ada kaedah yang lebih baik, bahagian kod ini boleh diganti. Berdasarkan reka bentuk yang mudah ini, menyesuaikan julat penjanaan nombor rawak dan beberapa pekali dalam kod boleh mempengaruhi kesan data yang dihasilkan.

• Pengesahan data Data K-line yang dihasilkan juga perlu diuji untuk rasional, untuk memeriksa sama ada harga pembukaan yang tinggi dan harga penutupan yang rendah melanggar definisi, dan untuk memeriksa kesinambungan data K-line.

Backtesting System Random Ticker Generator

import _thread
import json
import math
import csv
import random
import os
import datetime as dt
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse

arrTrendType = ["down", "slow_up", "sharp_down", "sharp_up", "narrow_range", "wide_range", "neutral_random"]

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):
        global filePathForCSV, pround, vround, ct
        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)            
            
            eid = dictParam["eid"]
            symbol = dictParam["symbol"]
            arrCurrency = symbol.split(".")[0].split("_")
            baseCurrency = arrCurrency[0]
            quoteCurrency = arrCurrency[1]
            fromTS = int(dictParam["from"]) * int(1000)
            toTS = int(dictParam["to"]) * int(1000)
            priceRatio = math.pow(10, int(pround))
            amountRatio = math.pow(10, int(vround))

            data = {
                "detail": {
                    "eid": eid,
                    "symbol": symbol,
                    "alias": symbol,
                    "baseCurrency": baseCurrency,
                    "quoteCurrency": quoteCurrency,
                    "marginCurrency": quoteCurrency,
                    "basePrecision": vround,
                    "quotePrecision": pround,
                    "minQty": 0.00001,
                    "maxQty": 9000,
                    "minNotional": 5,
                    "maxNotional": 9000000,
                    "priceTick": 10 ** -pround,
                    "volumeTick": 10 ** -vround,
                    "marginLevel": 10,
                    "contractType": ct
                },
                "schema" : ["time", "open", "high", "low", "close", "vol"],
                "data" : []
            }
            
            listDataSequence = []
            with open(filePathForCSV, "r") as f:
                reader = csv.reader(f)
                header = next(reader)
                headerIsNoneCount = 0
                if len(header) != len(data["schema"]):
                    Log("The CSV file format is incorrect, the number of columns is different, please check!", "#FF0000")
                    return 
                for ele in header:
                    for i in range(len(data["schema"])):
                        if data["schema"][i] == ele or ele == "":
                            if ele == "":
                                headerIsNoneCount += 1
                            if headerIsNoneCount > 1:
                                Log("The CSV file format is incorrect, please check!", "#FF0000")
                                return 
                            listDataSequence.append(i)
                            break
                
                while True:
                    record = next(reader, -1)
                    if record == -1:
                        break
                    index = 0
                    arr = [0, 0, 0, 0, 0, 0]
                    for ele in record:
                        arr[listDataSequence[index]] = int(ele) if listDataSequence[index] == 0 else (int(float(ele) * amountRatio) if listDataSequence[index] == 5 else int(float(ele) * priceRatio))
                        index += 1
                    data["data"].append(arr)            
            Log("data.detail: ", data["detail"], "Respond to backtesting system requests.")
            self.wfile.write(json.dumps(data).encode())
        except BaseException as e:
            Log("Provider do_GET error, e:", e)
        return 

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")

class KlineGenerator:
    def __init__(self, start_time, end_time, interval):
        self.start_time = dt.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
        self.end_time = dt.datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
        self.interval = self._parse_interval(interval)
        self.timestamps = self._generate_time_series()

    def _parse_interval(self, interval):
        unit = interval[-1]
        value = int(interval[:-1])

        if unit == "m":
            return value * 60
        elif unit == "h":
            return value * 3600
        elif unit == "d":
            return value * 86400
        else:
            raise ValueError("Unsupported K-line period, please use 'm', 'h', or 'd'.")

    def _generate_time_series(self):
        timestamps = []
        current_time = self.start_time
        while current_time <= self.end_time:
            timestamps.append(int(current_time.timestamp() * 1000))
            current_time += dt.timedelta(seconds=self.interval)
        return timestamps

    def generate(self, initPrice, trend_type="neutral", volatility=1):
        data = []
        current_price = initPrice
        angle = 0
        for timestamp in self.timestamps:
            angle_radians = math.radians(angle % 360)
            cos_value = math.cos(angle_radians)

            if trend_type == "down":
                upFactor = random.uniform(0, 0.5)
                change = random.uniform(-0.5, 0.5 * upFactor) * volatility * random.uniform(1, 3)
            elif trend_type == "slow_up":
                downFactor = random.uniform(0, 0.5)
                change = random.uniform(-0.5 * downFactor, 0.5) * volatility * random.uniform(1, 3)
            elif trend_type == "sharp_down":
                upFactor = random.uniform(0, 0.5)
                change = random.uniform(-10, 0.5 * upFactor) * volatility * random.uniform(1, 3)
            elif trend_type == "sharp_up":
                downFactor = random.uniform(0, 0.5)
                change = random.uniform(-0.5 * downFactor, 10) * volatility * random.uniform(1, 3)
            elif trend_type == "narrow_range":
                change = random.uniform(-0.2, 0.2) * volatility * random.uniform(1, 3)
            elif trend_type == "wide_range":
                change = random.uniform(-3, 3) * volatility * random.uniform(1, 3)
            else:
                change = random.uniform(-0.5, 0.5) * volatility * random.uniform(1, 3)

            change = change + cos_value * random.uniform(-0.2, 0.2) * volatility
            open_price = current_price
            high_price = open_price + random.uniform(0, abs(change))
            low_price = max(open_price - random.uniform(0, abs(change)), random.uniform(0, open_price))
            close_price = open_price + change if open_price + change < high_price and open_price + change > low_price else random.uniform(low_price, high_price)

            if (high_price >= open_price and open_price >= close_price and close_price >= low_price) or (high_price >= close_price and close_price >= open_price and open_price >= low_price):
                pass
            else:
                Log("Abnormal data:", high_price, open_price, low_price, close_price, "#FF0000")

            high_price = max(high_price, open_price, close_price)
            low_price = min(low_price, open_price, close_price)

            base_volume = random.uniform(1000, 5000)
            volume = base_volume * (1 + abs(change) * 0.2)

            kline = {
                "Time": timestamp,
                "Open": round(open_price, 2),
                "High": round(high_price, 2),
                "Low": round(low_price, 2),
                "Close": round(close_price, 2),
                "Volume": round(volume, 2),
            }
            data.append(kline)
            current_price = close_price
            angle += 1
        return data

    def save_to_csv(self, filename, data):
        with open(filename, mode="w", newline="") as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(["", "open", "high", "low", "close", "vol"])
            for idx, kline in enumerate(data):
                writer.writerow(
                    [kline["Time"], kline["Open"], kline["High"], kline["Low"], kline["Close"], kline["Volume"]]
                )
        
        Log("Current path:", os.getcwd())
        with open("data.csv", "r") as file:
            lines = file.readlines()
            if len(lines) > 1:
                Log("The file was written successfully. The following is part of the file content:")
                Log("".join(lines[:5]))
            else:
                Log("Failed to write the file, the file is empty!")

def main():
    Chart({})
    LogReset(1)
    
    try:
        # _thread.start_new_thread(createServer, (("localhost", 9090), ))
        _thread.start_new_thread(createServer, (("0.0.0.0", 9090), ))
        Log("Start the custom data source service thread, and the data is provided by the CSV file.", ", Address/Port: 0.0.0.0:9090", "#FF0000")
    except BaseException as e:
        Log("Failed to start custom data source service!")
        Log("error message:", e)
        raise Exception("stop")
    
    while True:
        cmd = GetCommand()
        if cmd:
            if cmd == "createRecords":
                Log("Generator parameters:", "Start time:", startTime, "End time:", endTime, "K-line period:", KLinePeriod, "Initial price:", firstPrice, "Type of volatility:", arrTrendType[trendType], "Volatility coefficient:", ratio)
                generator = KlineGenerator(
                    start_time=startTime,
                    end_time=endTime,
                    interval=KLinePeriod,
                )
                kline_data = generator.generate(firstPrice, trend_type=arrTrendType[trendType], volatility=ratio)
                generator.save_to_csv("data.csv", kline_data)
                ext.PlotRecords(kline_data, "%s_%s" % ("records", KLinePeriod))
        LogStatus(_D())
        Sleep(2000)

Amalan dalam Sistem Ujian Belakang

1. Buat contoh strategi di atas, sesuaikan parameter, dan jalankan.
2. Perdagangan langsung (contoh strategi) perlu dijalankan pada docker yang digunakan di pelayan, ia memerlukan IP rangkaian awam, supaya sistem backtesting dapat mengaksesnya dan mendapatkan data.
3. Klik butang interaksi, dan strategi akan mula menjana data ticker rawak secara automatik.

4. Data yang dihasilkan akan dipaparkan pada carta untuk pemerhatian yang mudah, dan data akan direkodkan dalam fail data.csv tempatan.

5. Sekarang kita boleh menggunakan data yang dihasilkan secara rawak ini dan menggunakan mana-mana strategi untuk backtesting:

/*backtest
start: 2024-10-01 08:00:00
end: 2024-10-31 08:55:00
period: 1h
basePeriod: 1h
exchanges: [{"eid":"Futures_Binance","currency":"BTC_USDT","feeder":"http://xxx.xxx.xxx.xxx:9090"}]
args: [["ContractType","quarter",358374]]
*/

Mengikut maklumat di atas, sesuaikan dan sesuaikan.http://xxx.xxx.xxx.xxx:9090adalah alamat IP pelayan dan pelabuhan terbuka strategi penjanaan ticker rawak. Ini adalah sumber data tersuai, yang boleh didapati di bahagian Sumber Data Tersuai dalam dokumen API platform.

6. Selepas sistem backtest menetapkan sumber data, kita boleh menguji data pasaran rawak:

Pada masa ini, sistem backtest diuji dengan data simulasi fabricated kami. Menurut data dalam carta ticker semasa backtest, data dalam carta perdagangan langsung yang dihasilkan oleh pasaran rawak dibandingkan. Waktu: 17:00 pada 16 Oktober 2024, dan data adalah sama.

7. Oh, ya, saya hampir lupa menyebutnya! Sebab mengapa program Python ini penjana ticker rawak mencipta perdagangan langsung adalah untuk memudahkan demonstrasi, operasi, dan paparan data K-line yang dihasilkan.

Kod sumber strategi:Backtesting System Random Ticker Generator

Terima kasih atas sokongan dan bacaan anda.