ہاتھ سے آپ کو کسٹم ڈیٹا ماخذ کی کارکردگی کی جانچ پڑتال کرنے کے لئے مارکیٹنگ کلکٹر کو اپ گریڈ کرنے کے لئے سکھاتا ہے

ہاتھ سے آپ کو کسٹم ڈیٹا ماخذ کی کارکردگی کی جانچ پڑتال کرنے کے لئے مارکیٹنگ کلکٹر کو اپ گریڈ کرنے کے لئے سکھاتا ہے

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

اس طرح ہم "بزنس کلکٹر" کو اپ گریڈ کر سکتے ہیں! تاکہ بزنس کلکٹر بیک وقت اپنی مرضی کے مطابق ڈیٹا ماخذ کے طور پر ریٹرننگ سسٹم کو ڈیٹا فراہم کرسکے۔

اگر ضرورت ہو تو، کام کریں!

تیار

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

• mongodb ڈیٹا بیس

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

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

  • پیمونگو
  • http
  • urllib

• نگہبان ایک مقتدر کو کوٹیفیکیشن ٹریڈنگ پلیٹ فارم چلانے والے کے لئے ایک ٹرسٹ۔

"بندوق جمع کرنے والے" کو تبدیل کریں

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

RecordsCollector (اپ گریڈ کردہ اپنی مرضی کے مطابق ڈیٹا ماخذ کی سہولت)

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

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

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

اس طرح، ہم اپنے روبوٹ کو وی پی ایس پر K لائن کے اعداد و شمار کو جمع کرنے کی اجازت دے سکتے ہیں، اور ہم ان اعداد و شمار کو براہ راست ریٹرننگ سسٹم میں حاصل کرسکتے ہیں. اس کے علاوہ، آپ کو مزید توسیع کی جا سکتی ہے، جیسے کہ ریئل ڈسک کی سطح پر اپنی مرضی کے مطابق اعداد و شمار کے ذرائع کی جانچ پڑتال، کثیر اقسام، کثیر مارکیٹ ڈیٹا جمع کرنے اور دیگر خصوصیات کی حمایت.

خوش آمدید۔