Metode pengujian strategi berdasarkan generator pasar acak

[TOC]

Pengantar

Penemu platform perdagangan kuantitatif adalah sebuah sistem yang terus-menerus diupgrade dari fitur dasar untuk meningkatkan fungsionalitas dan mengoptimalkan kinerja. Dengan perkembangan platform, sistem ini terus-menerus dioptimalkan.

Kebutuhan

Dalam bidang perdagangan kuantitatif, pengembangan strategi dan pengoptimalan tidak dapat dipisahkan dari verifikasi data pasar nyata. Namun, dalam aplikasi praktis, karena kompleksitas dan variasi lingkungan pasar, ketergantungan pada data sejarah untuk melakukan retrospeksi mungkin tidak cukup, seperti kurangnya cakupan pasar ekstrim atau skenario khusus. Oleh karena itu, merancang generator tren acak yang efisien menjadi alat yang efektif bagi pengembang strategi kuantitatif.

Ketika kita ingin membuat strategi yang dapat menelusuri data historis di suatu bursa, mata uang tertentu, kita dapat menggunakan sumber data resmi dari FMZ untuk melakukan pengujian kembali. Kadang-kadang kita juga ingin melihat bagaimana strategi itu akan berkinerja jika di pasar yang sama sekali tidak dikenal, saat ini kita dapat membuat data untuk menguji strategi.

Penggunaan data pasar acak berarti:

• 1. Mengevaluasi Strateginya Generator pasar acak dapat menciptakan berbagai skenario pasar yang mungkin terjadi, termasuk pasar yang berfluktuasi ekstrem, rendah fluktuasi, pasar tren, dan pasar yang bergejolak. Memeriksa strategi dalam lingkungan analog ini dapat membantu menilai apakah strategi tersebut stabil dalam berbagai kondisi pasar.

  Apakah strategi dapat beradaptasi dengan tren dan pergeseran goyah? Apakah strategi akan mengalami kerugian besar dalam pasar ekstrim?

• 2. Mengidentifikasi potensi kelemahan strategi Dengan mensimulasikan beberapa situasi pasar yang tidak biasa (misalnya, insiden Black Swan yang diandaikan), strategi dapat menemukan kelemahan potensial dan disempurnakan.

  Apakah strategi akan terlalu bergantung pada struktur pasar tertentu? Apakah ada risiko parameter yang terlalu pas?

• 3. Optimasi Parameter Strategi Data yang dihasilkan secara acak memberikan lingkungan pengujian yang lebih beragam untuk penyesuaian parameter strategi, tanpa harus sepenuhnya bergantung pada data historis. Hal ini memungkinkan untuk menemukan rentang parameter strategi yang lebih komprehensif, menghindari keterbatasan pada pola pasar tertentu dalam data historis.
• 4. Data sejarah yang tidak lengkap Dalam beberapa pasar (misalnya pasar berkembang atau pasar perdagangan mata uang kecil), data historis mungkin tidak cukup untuk mencakup semua kondisi pasar yang mungkin. Generator pasar acak dapat memberikan sejumlah besar data tambahan untuk membantu melakukan pengujian yang lebih komprehensif.
• 5. Pengembangan Iterasi Cepat Dengan menggunakan data acak untuk pengujian cepat, iterasi pengembangan strategi dapat dipercepat, tanpa harus bergantung pada pasar real-time atau pembersihan dan pengelompokan data yang memakan waktu.

Namun, strategi evaluasi yang rasional juga diperlukan, dan untuk data pasar yang dihasilkan secara acak, perhatikan:

• 1, Meskipun generator pasar acak berguna, artinya tergantung pada kualitas data yang dihasilkan dan desain skenario target:
• Logika generasi harus mendekati pasar nyata: jika pasar yang dihasilkan secara acak sepenuhnya lepas dari realitas, hasil pengujian mungkin tidak memiliki nilai referensi. Misalnya, dapat digabungkan dengan karakteristik statistik pasar yang sebenarnya (seperti distribusi volatilitas, rasio tren) untuk merancang generator.
• 3. Tidak dapat sepenuhnya menggantikan pengujian data nyata: data acak hanya dapat melengkapi pengembangan dan pengoptimalan strategi, dan strategi akhir masih perlu diverifikasi efektifitasnya dalam data pasar nyata.

Dengan begitu, bagaimana kita bisa membuat data yang mudah, cepat, dan mudah digunakan untuk membuat data yang mudah digunakan oleh sistem backtesting?

Pikiran Desain

Artikel ini dirancang untuk memberikan perhitungan generasi pasar acak yang relatif sederhana, tetapi sebenarnya ada berbagai macam algoritma simulasi, model data, dan teknik lain yang dapat diterapkan, karena diskusi terbatas tidak menggunakan metode simulasi data yang sangat rumit.

Kami menulis sebuah program dalam bahasa Python yang menggunakan fitur sumber data kustom dari sistem penelusuran platform.

• 1, secara acak menghasilkan seperangkat data K-line yang ditulis ke dalam file CSV, sehingga data yang dihasilkan dapat disimpan sebagai catatan.
• 2. kemudian membuat layanan untuk mendukung sumber data pada sistem penelusuran.
• 3, data K-line yang dihasilkan akan ditampilkan dalam grafik.

Untuk beberapa standar generasi data K-line, penyimpanan file, dll, kontrol parameter berikut dapat didefinisikan:

• Pola penciptaan data secara acak Untuk tipe fluktuasi yang meniru data K-line, hanya menggunakan angka acak yang sederhana untuk melakukan desain sederhana yang berbeda dari probabilitas positif negatif, yang mungkin tidak mencerminkan pola yang dibutuhkan ketika data yang dihasilkan tidak banyak. Jika ada cara yang lebih baik, bagian kode ini dapat diganti. Berdasarkan desain sederhana ini, penyesuaian rentang dan beberapa koefisien generasi angka acak dalam kode dapat mempengaruhi efek data yang dihasilkan.

• Pemeriksaan data Untuk data K-line yang dihasilkan juga perlu pengujian rasionalisasi, memeriksa apakah harga tinggi atau rendah melanggar definisi, memeriksa kontinuitas data K-line, dll.

Generator transaksi acak sistem retesting

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("自定义数据源服务接收到请求，self.path:", self.path, "query 参数：", 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("CSV文件格式有误，列数不同，请检查！", "#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("CSV文件格式有误，请检查！", "#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"], "响应回测系统请求。")
            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("不支持的K线周期，请使用 'm', 'h', 或 '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("异常数据：", 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("当前路径：", os.getcwd())
        with open("data.csv", "r") as file:
            lines = file.readlines()
            if len(lines) > 1:
                Log("文件写入成功，以下是文件内容的一部分：")
                Log("".join(lines[:5]))
            else:
                Log("文件写入失败，文件为空！")

def main():
    Chart({})
    LogReset(1)
    
    try:
        # _thread.start_new_thread(createServer, (("localhost", 9090), ))
        _thread.start_new_thread(createServer, (("0.0.0.0", 9090), ))
        Log("开启自定义数据源服务线程，数据由CSV文件提供。", ", 地址/端口：0.0.0.0:9090", "#FF0000")
    except BaseException as e:
        Log("启动自定义数据源服务失败！")
        Log("错误信息：", e)
        raise Exception("stop")
    
    while True:
        cmd = GetCommand()
        if cmd:
            if cmd == "createRecords":
                Log("生成器参数：", "起始时间：", startTime, "结束时间：", endTime, "K线周期：", KLinePeriod, "初始价格：", firstPrice, "波动类型：", arrTrendType[trendType], "波动性系数：", 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)

Praktik dalam sistem retesting

1, membuat contoh kebijakan di atas, mengkonfigurasi parameter, dan menjalankan. 2, Real Disk (instansi kebijakan) harus berjalan pada host yang ditempatkan di server, karena harus ada IP publik yang dapat diakses oleh sistem retargeting untuk mengambil data. 3. Klik tombol interaksi, dan strategi akan secara otomatis mulai menghasilkan data pasar acak.

4、生成好的数据会显示在图表上，方便观察，同时数据会记录在本地的data.csv文件

Jadi kita bisa menggunakan data yang dihasilkan secara acak ini untuk melakukan pengukuran ulang secara acak menggunakan strategi.

/*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]]
*/

Perangkat lunak ini dapat digunakan untuk mengkonfigurasikan dan mengkonfigurasikan aplikasi.http://xxx.xxx.xxx.xxx:9090Ini adalah alamat IP server dan port yang terbuka untuk generasi kebijakan disk secara acak. Ini adalah sumber data kustom, yang dapat Anda tanyakan pada bagian sumber data kustom dalam dokumentasi API platform.

6. Dengan pengaturan sumber data yang baik, Anda dapat menguji data pasar acak.

Pada saat ini, sistem retesting adalah untuk menguji data analog dari kerucut kami. Berdasarkan data dalam grafik pasar pada saat retesting, dibandingkan dengan data dalam grafik pasar yang dihasilkan secara acak, waktu: 16 Oktober 2024 pukul 17:00, data yang sama.

7. Oh ya, hampir lupa! Program python ini membuat sebuah cakram nyata untuk mendemonstrasikan, mengoperasikan, dan menampilkan data K-line yang dihasilkan.

Kode sumber strategi:Generator transaksi acak sistem retesting

Terima kasih atas dukungannya dan pembacanya.