مارکیٹ کلیکٹر کے بیکٹیسٹ کسٹم ڈیٹا سورس فنکشن کو اپ گریڈ کرنے کے لیے آپ کو مرحلہ وار سکھاتے ہیں۔

پچھلا مضمونآپ کو ایک مارکیٹ کلیکٹر کو لاگو کرنے کے لئے قدم بہ قدم سکھائیںہم نے مل کر مارکیٹ کا ڈیٹا اکٹھا کرنے کے لیے ایک روبوٹ پروگرام نافذ کیا ہے کہ ہم جمع کردہ مارکیٹ ڈیٹا کو کیسے استعمال کرتے ہیں؟ بلاشبہ، یہ بیک ٹیسٹنگ سسٹم کے لیے استعمال ہوتا ہے، یہاں موجد کے مقداری تجارتی پلیٹ فارم کے بیک ٹیسٹنگ سسٹم کے کسٹم ڈیٹا سورس فنکشن پر انحصار کرتے ہوئے، ہم جمع کیے گئے ڈیٹا کو بیک ٹیسٹنگ سسٹم کے ڈیٹا سورس کے طور پر استعمال کر سکتے ہیں، تاکہ ہم بیک ٹیسٹنگ سسٹم بنا سکتا ہے ٹیسٹنگ سسٹم کسی بھی مارکیٹ میں لاگو کیا جا سکتا ہے جہاں ہم تاریخی ڈیٹا کو بیک ٹیسٹ کرنا چاہتے ہیں۔

لہذا، ہم “مارکیٹ کلیکٹر” کو اپ گریڈ کر سکتے ہیں! بیک ٹیسٹنگ سسٹم کو ڈیٹا فراہم کرنے کے لیے مارکیٹ کلیکٹر کو حسب ضرورت ڈیٹا سورس کے طور پر بھی استعمال کیا جا سکتا ہے۔

اگر آپ کو ضرورت ہے تو کارروائی کریں!

تیار کریں۔

تیاری آخری مضمون سے مختلف ہیں، میں نے اپنے مقامی MAC کمپیوٹر پر میزبان پروگرام چلایا، mongodb ڈیٹا بیس انسٹال کیا اور ڈیٹا بیس سروس شروع کی۔ اس بار ہم نے آپریٹنگ ماحول کو VPS میں تبدیل کیا اور اپنے پروگرام کو چلانے کے لیے علی بابا کلاؤڈ لینکس سرور کا استعمال کیا۔

• mongodb ڈیٹا بیس

جیسا کہ پچھلے مضمون میں ہے، آپ کو اس ڈیوائس پر MongoDB ڈیٹا بیس انسٹال کرنے کی ضرورت ہے جہاں مارکیٹ کلیکٹر پروگرام چلتا ہے اور سروس شروع کرنا ہے۔ یہ بنیادی طور پر ایک MAC کمپیوٹر پر انسٹال کرنے کے مترادف ہے۔

• Python 3 انسٹال کریں۔ پروگرام Python 3 استعمال کرتا ہے۔ نوٹ کریں کہ کچھ لائبریریاں استعمال ہوتی ہیں اور اگر دستیاب نہ ہوں تو انسٹال کرنے کی ضرورت ہے۔

  • pymongo
  • http
  • urllib

• میزبان بس موجد مقداری تجارتی پلیٹ فارم کا ایک نگران چلائیں۔

“مارکیٹ انفارمیشن کلیکٹر” میں ترمیم کریں

مارکیٹ کلیکٹرریکارڈ جمع کرنے والا (ٹیوٹوریل)یہ حکمت عملی۔ آئیے اس میں کچھ تبدیلیاں کرتے ہیں: اس سے پہلے کہ پروگرام ڈیٹا اکٹھا کرنے کے دوران لوپ میں داخل ہو، ایک ملٹی تھریڈڈ لائبریری کا استعمال موجد کے مقداری تجارتی پلیٹ فارم بیک ٹیسٹنگ سسٹم سے ڈیٹا کی درخواستوں کو سننے کے لیے بیک وقت ایک سروس شروع کرنے کے لیے کیا جاتا ہے۔ (کچھ دیگر تفصیلات کو نظر انداز کیا جا سکتا ہے)

Recordscollecter (کسٹم ڈیٹا سورس فنکشن فراہم کرنے کے لیے اپ گریڈ کیا گیا)

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("自定义数据源服务接收到请求，self.path:", self.path, "query 参数：", dictParam)
            
            # 目前回测系统只能从列表中选择交易所名称，在添加自定义数据源时，设置为币安，即：Binance
            exName = exchange.GetName()                                     
            # 注意，period为底层K线周期
            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)
            
            
            # 连接数据库
            Log("连接数据库服务，获取数据，数据库：", exName, "表：", tabName)
            myDBClient = pymongo.MongoClient("mongodb://localhost:27017")
            ex_DB = myDBClient[exName]
            exRecords = ex_DB[tabName]
            
            
            # 要求应答的数据
            data = {
                "schema" : ["time", "open", "high", "low", "close", "vol"],
                "data" : []
            }
            
            # 构造查询条件：大于某个值{'age': {'$gt': 20}} 小于某个值{'age': {'$lt': 20}}
            dbQuery = {"$and":[{'Time': {'$gt': fromTS}}, {'Time': {'$lt': toTS}}]}
            Log("查询条件：", dbQuery, "查询条数：", exRecords.find(dbQuery).count(), "数据库总条数：", exRecords.find().count())
            
            for x in exRecords.find(dbQuery).sort("Time"):
                # 需要根据请求参数round和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, "响应回测系统请求。")
            # 写入数据应答
            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("收集", exName, "交易所的K线数据，", "K线周期:", period, "秒")
    
    # 连接数据库服务，服务地址 mongodb://127.0.0.1:27017 具体看服务器上安装的mongodb设置
    Log("连接托管者所在设备mongodb服务，mongodb://localhost:27017")
    myDBClient = pymongo.MongoClient("mongodb://localhost:27017")   
    # 创建数据库
    ex_DB = myDBClient[exName]
    
    # 打印目前数据库表
    collist = ex_DB.list_collection_names()
    Log("mongodb ", exName, " collist:", collist)
    
    # 检测是否删除表
    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, "删除：", dropName)
                ret = tab.drop()
                collist = ex_DB.list_collection_names()
                if dropName in collist:
                    Log(dropName, "删除失败")
                else :
                    Log(dropName, "删除成功")
    
    # 开启一个线程，提供自定义数据源服务
    try:
        # _thread.start_new_thread(createServer, (("localhost", 9090), ))     # 本机测试
        _thread.start_new_thread(createServer, (("0.0.0.0", 9090), ))         # VPS服务器上测试
        Log("开启自定义数据源服务线程", "#FF0000")
    except BaseException as e:
        Log("启动自定义数据源服务失败！")
        Log("错误信息：", e)
        raise Exception("stop")
    
    # 创建records表
    ex_DB_Records = ex_DB["%s_%d" % ("records", period)]
    Log("开始收集", exName, "K线数据", "周期：", period, "打开（创建）数据库表：", "%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:
            # 首次写入所有BAR数据
            for i in range(len(r) - 1):
                bar = r[i]
                # 逐根写入，需要判断当前数据库表中是否已经有该条数据，基于时间戳检测，如果有该条数据，则跳过，没有则写入
                retQuery = ex_DB_Records.find({"Time": bar["Time"]})
                if retQuery.count() > 0:
                    continue
                
                # 写入bar到数据库表
                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]
            # 写入数据前检测，数据是否已经存在，基于时间戳检测
            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)
        # 增加画图展示
        ext.PlotRecords(r, "%s_%d" % ("records", period))
        Sleep(10000)
        

ٹیسٹ

روبوٹ کو ترتیب دیں۔

روبوٹ چلائیں اور بازار جمع کرنے والے کو چلائیں۔

ایک ٹیسٹ کی حکمت عملی کھولیں اور بیک ٹیسٹ کریں، جیسے کہ یہ بیک ٹیسٹ حکمت عملی، اور اس کی جانچ کریں۔

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

بیک ٹیسٹنگ کے اختیارات کو کنفیگر کریں اور ایکسچینج کو بائننس پر سیٹ کریں کیونکہ اس وقت آپ بائننس کو ترتیب دینے کے لیے ایک ایکسچینج کا استعمال کر سکتے ہیں۔ wexApp سمولیشن ڈسک کا۔

بیک ٹیسٹنگ سسٹم کے ذریعہ تیار کردہ چارٹ کا مارکیٹ کلیکٹر کی بنیاد پر حسب ضرورت ڈیٹا سورس کے طور پر اور wexApp ایکسچینج پیج پر 1 گھنٹے کے کینڈل اسٹک چارٹ کا موازنہ کریں تاکہ یہ معلوم ہو سکے کہ آیا وہ ایک جیسے ہیں۔

اس طرح، VPS پر روبوٹ خود سے K-line ڈیٹا اکٹھا کر سکتا ہے، اور ہم جمع کردہ ڈیٹا کو کسی بھی وقت حاصل کر سکتے ہیں اور بیک ٹیسٹنگ سسٹم میں براہ راست بیک ٹیسٹ کر سکتے ہیں۔ یہ سمندر میں صرف ایک قطرہ ہے، ماہرین اس پر توسیع جاری رکھ سکتے ہیں، مثال کے طور پر، ریئل ٹائم بیک ٹیسٹنگ، ملٹی ورائٹی اور ملٹی مارکیٹ ڈیٹا اکٹھا کرنے اور دیگر افعال کی حمایت کرنا۔

ایک پیغام چھوڑنے میں خوش آمدید۔