[TOC]

ভূমিকা

উদ্ভাবক পরিমাণগত ট্রেডিং প্ল্যাটফর্মের ব্যাকটেস্টিং সিস্টেমটি একটি ব্যাকটেস্টিং সিস্টেম যা ক্রমাগত পুনরাবৃত্তভাবে আপডেট করা হয় এবং প্রাথমিক বেসিক ব্যাকটেস্টিং ফাংশন থেকে এটি ধীরে ধীরে ফাংশন যোগ করে এবং কর্মক্ষমতা অপ্টিমাইজ করে। প্ল্যাটফর্মের বিকাশের সাথে সাথে, ব্যাকটেস্টিং সিস্টেমটি অপ্টিমাইজ করা এবং আপগ্রেড করা অব্যাহত থাকবে আজ আমরা ব্যাকটেস্টিং সিস্টেমের উপর ভিত্তি করে একটি বিষয় নিয়ে আলোচনা করব: “এলোমেলো বাজারের অবস্থার উপর ভিত্তি করে কৌশল পরীক্ষা।”

প্রয়োজন

পরিমাণগত ট্রেডিংয়ের ক্ষেত্রে, কৌশলগুলির বিকাশ এবং অপ্টিমাইজেশন বাস্তব বাজারের ডেটা যাচাইকরণ থেকে অবিচ্ছেদ্য। যাইহোক, ব্যবহারিক প্রয়োগে, জটিল এবং পরিবর্তনশীল বাজার পরিবেশের কারণে, ব্যাকটেস্টিংয়ের জন্য ঐতিহাসিক তথ্যের উপর নির্ভর করার ত্রুটিগুলি থাকতে পারে, যেমন চরম বাজারের অবস্থার কভারেজের অভাব বা বিশেষ পরিস্থিতি। অতএব, একটি দক্ষ এলোমেলো বাজার জেনারেটর ডিজাইন করা পরিমাণগত কৌশল বিকাশকারীদের জন্য একটি কার্যকর হাতিয়ার হয়ে উঠেছে।

যখন আমাদের একটি নির্দিষ্ট বিনিময় বা মুদ্রায় ঐতিহাসিক ডেটা ব্যাকটেস্ট করার কৌশলের প্রয়োজন হয়, তখন আমরা ব্যাকটেস্টিংয়ের জন্য FMZ প্ল্যাটফর্মের অফিসিয়াল ডেটা উৎস ব্যবহার করতে পারি। কখনও কখনও আমরা দেখতে চাই যে কৌশলটি একটি সম্পূর্ণ “অপরিচিত” বাজারে কীভাবে কাজ করবে এই সময়ে, আমরা কৌশলটি পরীক্ষা করার জন্য কিছু ডেটা “মেক আপ” করতে পারি।

এলোমেলো বাজার তথ্য ব্যবহারের তাৎপর্য হল:

• 1. আপনার কৌশলের দৃঢ়তা মূল্যায়ন করুন র্যান্ডম মার্কেট জেনারেটর বিভিন্ন সম্ভাব্য বাজারের পরিস্থিতি তৈরি করতে পারে, যার মধ্যে চরম অস্থিরতা, কম অস্থিরতা, ট্রেন্ডিং মার্কেট এবং অস্থির বাজার। এই সিমুলেটেড পরিবেশে পরীক্ষার কৌশলগুলি বিভিন্ন বাজারের অবস্থার অধীনে তাদের কর্মক্ষমতা স্থিতিশীল হবে কিনা তা মূল্যায়ন করতে সহায়তা করতে পারে। যেমন:

কৌশল প্রবণতা এবং সুইচ সুইচ মানিয়ে নিতে পারে? কৌশলটি কি চরম বাজারের পরিস্থিতিতে উল্লেখযোগ্য ক্ষতির সম্মুখীন হবে?

• 2. আপনার কৌশল সম্ভাব্য দুর্বলতা চিহ্নিত করুন কিছু অস্বাভাবিক বাজার পরিস্থিতি (যেমন একটি অনুমানমূলক কালো রাজহাঁস ইভেন্ট) অনুকরণ করে, কৌশলটির সম্ভাব্য দুর্বলতাগুলি আবিষ্কার এবং উন্নত করা যেতে পারে। যেমন:

কৌশলটি কি একটি নির্দিষ্ট বাজার কাঠামোর উপর অত্যধিক নির্ভরশীল হবে? পরামিতি overfitting একটি ঝুঁকি আছে?

• 1. কৌশল পরামিতি অপ্টিমাইজ করুন এলোমেলোভাবে তৈরি করা ডেটা সম্পূর্ণরূপে ঐতিহাসিক ডেটার উপর নির্ভর না করে নীতি প্যারামিটার টিউনিংয়ের জন্য আরও বৈচিত্র্যময় পরীক্ষার পরিবেশ প্রদান করে। এটি কৌশলটির জন্য পরামিতিগুলির আরও বিস্তৃত পরিসর খুঁজে পাওয়ার অনুমতি দেয় এবং ঐতিহাসিক ডেটাতে নির্দিষ্ট বাজারের প্যাটার্নগুলিতে সীমাবদ্ধ থাকা এড়িয়ে যায়।
• 1. ঐতিহাসিক তথ্যের শূন্যস্থান পূরণ করুন কিছু বাজারে (যেমন উদীয়মান বাজার বা ছোট মুদ্রা লেনদেন বাজার), ঐতিহাসিক তথ্য সমস্ত সম্ভাব্য বাজার পরিস্থিতি কভার করার জন্য যথেষ্ট নাও হতে পারে। একটি র্যান্ডম টিকার জেনারেটর আরও ব্যাপক পরীক্ষা পরিচালনা করতে সাহায্য করার জন্য প্রচুর পরিপূরক ডেটা সরবরাহ করতে পারে।
• 1. দ্রুত পুনরাবৃত্তিমূলক উন্নয়ন দ্রুত পরীক্ষার জন্য র্যান্ডম ডেটা ব্যবহার করা রিয়েল-টাইম বাজারের অবস্থা বা সময়-সাপেক্ষ ডেটা পরিষ্কার এবং সংস্থার উপর নির্ভর না করে কৌশল বিকাশের পুনরাবৃত্তিকে দ্রুত করতে পারে।

যাইহোক, এলোমেলোভাবে উত্পন্ন বাজার ডেটার জন্য কৌশলটি যুক্তিযুক্তভাবে মূল্যায়ন করা প্রয়োজন, দয়া করে নোট করুন:

• 1. যদিও র‍্যান্ডম মার্কেট জেনারেটর দরকারী, তবে এর তাৎপর্য উত্পন্ন ডেটার গুণমান এবং লক্ষ্য দৃশ্যের নকশার উপর নির্ভর করে:
• 2. প্রজন্মের যুক্তি বাস্তব বাজারের কাছাকাছি হওয়া প্রয়োজন: যদি এলোমেলোভাবে উৎপন্ন বাজার মূল্য বাস্তবতা থেকে সম্পূর্ণভাবে বিচ্ছিন্ন হয়, তাহলে পরীক্ষার ফলাফলের রেফারেন্স মানের অভাব হতে পারে। উদাহরণস্বরূপ, জেনারেটরটি প্রকৃত বাজারের পরিসংখ্যানগত বৈশিষ্ট্যগুলির সাথে একত্রে ডিজাইন করা যেতে পারে (যেমন অস্থিরতা বিতরণ, প্রবণতা অনুপাত)।
• 3. এটি বাস্তব ডেটা টেস্টিংকে সম্পূর্ণরূপে প্রতিস্থাপন করতে পারে না: র্যান্ডম ডেটা শুধুমাত্র কৌশলগুলির বিকাশ এবং অপ্টিমাইজেশনের পরিপূরক হতে পারে এবং চূড়ান্ত কৌশলটি এখনও বাস্তব বাজারের ডেটাতে এর কার্যকারিতা যাচাই করতে হবে।

যে সব বলে, কিভাবে আমরা কিছু তথ্য “তৈরি” করতে পারেন. ব্যাকটেস্টিং সিস্টেম ব্যবহারের জন্য আমরা কীভাবে সুবিধামত, দ্রুত এবং সহজে ডেটা “তৈরি” করতে পারি?

ডিজাইনের ধারণা

এই নিবন্ধটি কিছু ধারনা প্রবর্তন করার জন্য ডিজাইন করা হয়েছে এবং প্রকৃতপক্ষে, এখানে বিভিন্ন ধরণের সিমুলেশন অ্যালগরিদম, ডেটা মডেল এবং অন্যান্য প্রযুক্তি রয়েছে যা আলোচনার স্থান সীমিত, বিশেষ করে জটিল ডেটা সিমুলেশন পদ্ধতি ব্যবহার করা হবে না।

প্ল্যাটফর্ম ব্যাকটেস্টিং সিস্টেমের কাস্টম ডেটা সোর্স ফাংশনের সাথে মিলিত, আমরা একটি প্রোগ্রাম লিখতে পাইথন ভাষা ব্যবহার করি।

• 1. এলোমেলোভাবে কে-লাইন ডেটার একটি সেট তৈরি করুন এবং স্থায়ী রেকর্ডের জন্য এটিকে একটি CSV ফাইলে লিখুন, যাতে জেনারেট করা ডেটা রেকর্ড হিসাবে সংরক্ষণ করা যায়।
• 2. তারপর ব্যাকটেস্টিং সিস্টেমের জন্য ডেটা উত্স সমর্থন প্রদানের জন্য একটি পরিষেবা তৈরি করুন৷
• 3. চার্টে জেনারেট করা কে-লাইন ডেটা প্রদর্শন করুন।

কে-লাইন ডেটার কিছু প্রজন্মের মান, ফাইল স্টোরেজ ইত্যাদির জন্য, নিম্নলিখিত প্যারামিটার নিয়ন্ত্রণগুলি সংজ্ঞায়িত করা যেতে পারে:

• ডেটাতে এলোমেলোভাবে তৈরি প্যাটার্ন সিমুলেটেড কে-লাইন ডেটার ওঠানামার জন্য, এটি শুধুমাত্র একটি সাধারণ নকশা যা ধনাত্মক এবং নেতিবাচক র্যান্ডম সংখ্যার বিভিন্ন সম্ভাব্যতা ব্যবহার করে যখন তৈরি করা ডেটা বেশি না হয়, তখন এটি প্রয়োজনীয় বাজারের প্যাটার্নকে প্রতিফলিত করতে সক্ষম নাও হতে পারে। যদি একটি ভাল উপায় থাকে, আপনি কোডের এই অংশটি প্রতিস্থাপন করতে পারেন। এই সাধারণ নকশার উপর ভিত্তি করে, এলোমেলো সংখ্যা তৈরির পরিসর এবং কোডে কিছু সহগ সামঞ্জস্য করা উত্পন্ন ডেটা প্রভাবকে প্রভাবিত করতে পারে।

• ডেটা পরিদর্শন জেনারেট করা কে-লাইন ডেটাও যৌক্তিকতার জন্য পরীক্ষা করা প্রয়োজন, উচ্চ খোলার এবং কম ক্লোজিং মূল্য সংজ্ঞা লঙ্ঘন করে কিনা তা পরীক্ষা করতে এবং কে-লাইন ডেটার ধারাবাহিকতা পরীক্ষা করা ইত্যাদি।

ব্যাকটেস্টিং সিস্টেম র্যান্ডম মার্কেট জেনারেটর

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. আসল ডিস্ক (কৌশল উদাহরণ) সার্ভারে স্থাপন করা হোস্টে চালানো দরকার, কারণ এটির একটি পাবলিক আইপি প্রয়োজন যাতে ব্যাকটেস্ট সিস্টেম এটি অ্যাক্সেস করতে পারে এবং ডেটা পেতে পারে। ৩. ইন্টারেক্টিভ বোতামে ক্লিক করুন এবং কৌশলটি স্বয়ংক্রিয়ভাবে এলোমেলো বাজার তথ্য তৈরি করা শুরু করবে।

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. ওহ হ্যাঁ, আমি এটা বলতে প্রায় ভুলে গেছি! এই র্যান্ডম মার্কেট জেনারেটরের পাইথন প্রোগ্রামটি জেনারেট করা কে-লাইন ডেটা প্রদর্শন, অপারেশন এবং প্রদর্শনের সুবিধার্থে একটি বাস্তব বাজার তৈরি করে। প্রকৃত অ্যাপ্লিকেশনে, আপনি একটি পৃথক পাইথন স্ক্রিপ্ট লিখতে পারেন যাতে আপনাকে একটি বাস্তব ডিস্ক চালাতে না হয়।

কৌশল সোর্স কোড:ব্যাকটেস্টিং সিস্টেম র্যান্ডম মার্কেট জেনারেটর

আপনার সমর্থন এবং পড়ার জন্য ধন্যবাদ.