[TOC]

प्रस्तावना

इन्वेंटर क्वांटिटेटिव ट्रेडिंग प्लेटफ़ॉर्म का बैकटेस्टिंग सिस्टम एक बैकटेस्टिंग सिस्टम है जो लगातार पुनरावृत्ति, अद्यतन और उन्नयन करता रहता है। शुरुआती बुनियादी बैकटेस्टिंग फ़ंक्शन से, यह धीरे-धीरे फ़ंक्शन जोड़ता है और प्रदर्शन को अनुकूलित करता है। जैसे-जैसे प्लेटफ़ॉर्म विकसित होता जाएगा, बैकटेस्टिंग सिस्टम को ऑप्टिमाइज़ और अपग्रेड किया जाता रहेगा। आज हम बैकटेस्टिंग सिस्टम पर आधारित एक विषय पर चर्चा करेंगे: “यादृच्छिक बाज़ार स्थितियों पर आधारित रणनीति परीक्षण”।

ज़रूरत

मात्रात्मक व्यापार के क्षेत्र में, रणनीतियों के विकास और अनुकूलन को वास्तविक बाजार डेटा के सत्यापन से अलग नहीं किया जा सकता है। हालांकि, वास्तविक अनुप्रयोगों में, जटिल और बदलते बाजार परिवेश के कारण, बैकटेस्टिंग के लिए ऐतिहासिक डेटा पर निर्भर रहने में कमियां हो सकती हैं, जैसे कि चरम बाजार स्थितियों या विशेष परिदृश्यों के कवरेज की कमी। इसलिए, एक कुशल यादृच्छिक बाजार जनरेटर का डिजाइन मात्रात्मक रणनीति डेवलपर्स के लिए एक प्रभावी उपकरण बन जाता है।

जब हमें ऐतिहासिक डेटा का उपयोग करके किसी निश्चित एक्सचेंज या मुद्रा पर रणनीति का बैकटेस्ट करने की आवश्यकता होती है, तो हम बैकटेस्टिंग के लिए FMZ प्लेटफॉर्म के आधिकारिक डेटा स्रोत का उपयोग कर सकते हैं। कभी-कभी हम यह भी देखना चाहते हैं कि कोई रणनीति पूरी तरह से “अपरिचित” बाज़ार में कैसा प्रदर्शन करती है। इस समय, हम रणनीति का परीक्षण करने के लिए कुछ डेटा “गढ़” सकते हैं।

यादृच्छिक बाजार डेटा का उपयोग करने का महत्व यह है:

• 1. रणनीति की मजबूती का मूल्यांकन यादृच्छिक बाजार जनरेटर विभिन्न प्रकार के संभावित बाजार परिदृश्यों का निर्माण कर सकता है, जिनमें अत्यधिक अस्थिरता, कम अस्थिरता, ट्रेंडिंग बाजार और अस्थिर बाजार शामिल हैं। इन अनुरूपित वातावरणों में किसी रणनीति का परीक्षण करने से यह आकलन करने में मदद मिल सकती है कि विभिन्न बाजार स्थितियों में उसका प्रदर्शन स्थिर है या नहीं। उदाहरण के लिए:

क्या रणनीति प्रवृत्ति और आघात परिवर्तन के अनुकूल हो सकती है? क्या इस रणनीति के परिणामस्वरूप चरम बाजार स्थितियों में भारी नुकसान होगा?

• 2. अपनी रणनीति में संभावित कमजोरियों की पहचान करें कुछ असामान्य बाजार स्थितियों (जैसे काल्पनिक ब्लैक स्वान घटनाएं) का अनुकरण करके, रणनीति में संभावित कमजोरियों का पता लगाया जा सकता है और उनमें सुधार किया जा सकता है। उदाहरण के लिए:

क्या रणनीति किसी विशेष बाजार संरचना पर बहुत अधिक निर्भर करती है? क्या पैरामीटर्स के ओवरफिट होने का खतरा है?

• 1. रणनीति मापदंडों का अनुकूलन यादृच्छिक रूप से उत्पन्न डेटा, ऐतिहासिक डेटा पर पूरी तरह निर्भर हुए बिना, रणनीति पैरामीटर ट्यूनिंग के लिए अधिक विविध परीक्षण वातावरण प्रदान करता है। इससे रणनीति मापदंडों की अधिक व्यापक श्रृंखला उपलब्ध हो जाती है और ऐतिहासिक आंकड़ों में विशिष्ट बाजार पैटर्न तक सीमित होने से बचा जा सकता है।
• 1. ऐतिहासिक डेटा में अंतर को भरना कुछ बाजारों में (जैसे उभरते बाजार या छोटी मुद्राओं में व्यापार करने वाले बाजार), ऐतिहासिक डेटा सभी संभावित बाजार स्थितियों को कवर करने के लिए पर्याप्त नहीं हो सकता है। अधिक व्यापक परीक्षण की सुविधा के लिए रैंडमाइज़र बड़ी मात्रा में पूरक डेटा प्रदान कर सकता है।
• 1. तीव्र पुनरावृत्तीय विकास तीव्र परीक्षण के लिए यादृच्छिक डेटा का उपयोग करने से वास्तविक समय की बाजार स्थितियों या समय लेने वाली डेटा सफाई और संगठन पर निर्भर हुए बिना रणनीति विकास की पुनरावृत्ति में तेजी आ सकती है।

हालाँकि, रणनीति का तर्कसंगत मूल्यांकन करना भी आवश्यक है। यादृच्छिक रूप से उत्पन्न बाजार डेटा के लिए, कृपया ध्यान दें:

• 1. यद्यपि यादृच्छिक बाजार जनरेटर उपयोगी हैं, उनका महत्व उत्पन्न डेटा की गुणवत्ता और लक्ष्य परिदृश्य के डिजाइन पर निर्भर करता है:
• 2. उत्पादन तर्क वास्तविक बाजार के करीब होना चाहिए: यदि यादृच्छिक रूप से उत्पन्न बाजार की स्थितियां वास्तविकता से पूरी तरह से अलग हैं, तो परीक्षण के परिणामों में संदर्भ मूल्य का अभाव हो सकता है। उदाहरण के लिए, जनरेटर को वास्तविक बाजार सांख्यिकीय विशेषताओं (जैसे अस्थिरता वितरण, प्रवृत्ति अनुपात) के संयोजन में डिज़ाइन किया जा सकता है।
• 3. यह वास्तविक डेटा परीक्षण को पूरी तरह से प्रतिस्थापित नहीं कर सकता: यादृच्छिक डेटा केवल रणनीतियों के विकास और अनुकूलन को पूरक कर सकता है। अंतिम रणनीति को अभी भी वास्तविक बाजार डेटा में इसकी प्रभावशीलता को सत्यापित करने की आवश्यकता है।

इतना सब कहने के बाद, हम कुछ डेटा को कैसे “गढ़” सकते हैं। हम बैकटेस्टिंग प्रणाली में उपयोग के लिए सुविधाजनक, शीघ्र और आसानी से डेटा का निर्माण कैसे कर सकते हैं?

डिजाइन विचार

यह लेख चर्चा के लिए एक प्रारंभिक बिंदु प्रदान करने के लिए डिज़ाइन किया गया है और एक अपेक्षाकृत सरल यादृच्छिक बाजार निर्माण गणना प्रदान करता है। वास्तव में, कई प्रकार के सिमुलेशन एल्गोरिदम, डेटा मॉडल और अन्य तकनीकें हैं जिन्हें लागू किया जा सकता है। चर्चा के सीमित स्थान के कारण , हम विशेष रूप से जटिल डेटा सिमुलेशन विधियों का उपयोग नहीं करेंगे।

प्लेटफ़ॉर्म बैकटेस्टिंग सिस्टम के कस्टम डेटा स्रोत फ़ंक्शन को संयोजित करके, हमने पायथन में एक प्रोग्राम लिखा।

• 1. K-लाइन डेटा का एक सेट यादृच्छिक रूप से उत्पन्न करें और उन्हें स्थायी रिकॉर्डिंग के लिए CSV फ़ाइल में लिखें, ताकि उत्पन्न डेटा को सहेजा जा सके।
• 2. फिर बैकटेस्टिंग सिस्टम के लिए डेटा स्रोत समर्थन प्रदान करने के लिए एक सेवा बनाएं।
• 3. उत्पन्न K-लाइन डेटा को चार्ट में प्रदर्शित करें।

के-लाइन डेटा के कुछ उत्पादन मानकों और फ़ाइल भंडारण के लिए, निम्नलिखित पैरामीटर नियंत्रण परिभाषित किए जा सकते हैं:

• यादृच्छिक रूप से उत्पन्न डेटा पैटर्न के-लाइन डेटा के उतार-चढ़ाव प्रकार का अनुकरण करने के लिए, हम एक सरल डिज़ाइन बनाने के लिए बस सकारात्मक और नकारात्मक यादृच्छिक संख्याओं की विभिन्न संभावनाओं का उपयोग करते हैं। जब उत्पन्न डेटा बहुत अधिक नहीं होता है, तो यह आवश्यक बाजार पैटर्न को प्रतिबिंबित करने में सक्षम नहीं हो सकता है। यदि कोई बेहतर तरीका हो तो आप कोड के इस भाग को बदल सकते हैं। इस सरल डिजाइन के आधार पर, कोड में यादृच्छिक संख्या पीढ़ी सीमा और कुछ गुणांकों को समायोजित करने से उत्पन्न डेटा प्रभाव प्रभावित हो सकता है।

• डेटा सत्यापन उत्पन्न के-लाइन डेटा की तर्कसंगतता की भी जांच की जानी चाहिए, ताकि यह जांचा जा सके कि उच्च आरंभिक और निम्न समापन मूल्य परिभाषा का उल्लंघन करते हैं या नहीं, के-लाइन डेटा की निरंतरता की जांच की जा सके, आदि।

बैकटेस्टिंग सिस्टम रैंडम कोट जनरेटर

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)

बैकटेस्टिंग सिस्टम में अभ्यास

1. उपरोक्त नीति उदाहरण बनाएं, पैरामीटर कॉन्फ़िगर करें और इसे चलाएं।
2. वास्तविक बाज़ार (रणनीति उदाहरण) को सर्वर पर तैनात होस्ट पर चलाने की आवश्यकता होती है, क्योंकि बैकटेस्टिंग सिस्टम को उस तक पहुंचने और डेटा प्राप्त करने के लिए एक सार्वजनिक आईपी की आवश्यकता होती है।
3. इंटरैक्टिव बटन पर क्लिक करें और रणनीति स्वचालित रूप से यादृच्छिक बाजार डेटा उत्पन्न करना शुरू कर देगी।

1. उत्पन्न डेटा को आसान अवलोकन के लिए चार्ट पर प्रदर्शित किया जाएगा, और डेटा को स्थानीय data.csv फ़ाइल में रिकॉर्ड किया जाएगा

1. अब हम इस यादृच्छिक रूप से उत्पन्न डेटा का उपयोग कर सकते हैं और बैकटेस्टिंग के लिए किसी भी रणनीति का उपयोग कर सकते हैं

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

उपरोक्त जानकारी के अनुसार कॉन्फ़िगर करें और विशिष्ट समायोजन करें।http://xxx.xxx.xxx.xxx:9090यह सर्वर आईपी पता और यादृच्छिक बाजार पीढ़ी रणनीति वास्तविक डिस्क का खुला पोर्ट है। यह एक कस्टम डेटा स्रोत है। अधिक जानकारी के लिए आप प्लेटफ़ॉर्म API दस्तावेज़ में कस्टम डेटा स्रोत अनुभाग देख सकते हैं।

1. बैकटेस्ट सिस्टम द्वारा डेटा स्रोत सेट करने के बाद, आप यादृच्छिक बाज़ार डेटा का परीक्षण कर सकते हैं

इस बिंदु पर, बैकटेस्टिंग प्रणाली का परीक्षण हमारे “निर्मित” सिम्युलेटेड डेटा का उपयोग करके किया जाता है। बैकटेस्ट के दौरान मार्केट चार्ट में मौजूद डेटा के अनुसार, रैंडम मार्केट स्थितियों द्वारा उत्पन्न रियल-टाइम चार्ट में मौजूद डेटा की तुलना करें। समय 16 अक्टूबर 2024 को 17:00 बजे है। डेटा वही है।

1. ओह हाँ, मैं तो यह कहना ही भूल गया था! इस यादृच्छिक बाजार जनरेटर पायथन प्रोग्राम द्वारा वास्तविक बाजार बनाने का कारण उत्पन्न के-लाइन डेटा के प्रदर्शन, संचालन और प्रदर्शन को सुविधाजनक बनाना है। वास्तविक अनुप्रयोग में, आप एक स्वतंत्र पायथन स्क्रिप्ट लिख सकते हैं, इसलिए आपको वास्तविक डिस्क चलाने की आवश्यकता नहीं है।

रणनीति स्रोत कोड:बैकटेस्टिंग सिस्टम रैंडम कोट जनरेटर

आपके समर्थन और पढ़ने के लिए धन्यवाद।