高频交易中的tick数据是什么？如何接入tick数据（含代码示例）

金融交易世界中，获取准确及时的信息至关重要。抓住交易良机的关键在于掌握实时数据。数据更新越快，可发现的赚钱机会就越多。这也是为何在高频交易领域，tick数据备受重视。与传统的行情数据相比，tick数据提供了更细致的市场变化记录，为交易者提供了更全面的视角。

首先，让我们简单了解一下什么是Tick数据。

什么是Tick数据？

我们日常看到的K线行情数据是基于时间单位的，而tick数据则记录了更细致的维度，即每次价格变化都被记录。举例来说：假设一个股票在一分钟内变动了30次价格，那么在一分钟的行情数据中，你只会看到4个价格：

开盘价
收盘价
{BANNED}最佳高价
{BANNED}最佳低价

因为只有这4项数据，就足够绘制出一个蜡烛图:

O=开盘价，L={BANNED}最佳低价，H={BANNED}最佳高价，C=收盘价

而另外的26次价格变动，则被忽略了。

Tick数据与此不同，它会提供特定时间内的所有变化，即刚才提到的30次价格变动都会被记录。也就是说，你能够在一分钟内看到股票价格变动了30次。因此，tick数据也被称为高频数据。

高频数据可以帮助我们更好地了解市场行为和微观结构，同时也能够在非常短的时间范围内探索各种交易策略和假设。相反，对于许多应用场景来说，仅采样离散的低频数据并不能提供足够全面的市场分析。基于每日数据的研究和分析很可能会忽略大量的重要信息。

这些本该被忽略的、看似混乱的价格变动其实包含了非常多宝贵的信息。传统的行情数据是每分钟更新一次报价，而tick数据真正做到了实时更新，即每发生一笔交易都记录在内，它的更新时间是不固定的，完全随机的，因为你无法知道下一次交易发生在什么时候。因此，我们可以从交易之间的时间间隔中推测出目前市场的波动性、流动性等市场趋势。

如果你拥有长时间的，比如说5年的tick历史数据库，你就可以用来进行回测。

Tick数据的用途

高频数据已经广泛运用于量化交易的各个环节，其中比较经典的应用场景是回溯测试，即所谓的回测（Backtesting）。

回测是量化交易里非常重要的环节，当需要验证一套交易策略是否有效时，{BANNED}最佳简单的方法就是将其应用到真实的历史行情中，观察整个策略在其中的收益和{BANNED}最佳大回撤。回溯测试是基于这样一种理念，即如果我这套策略在过去表现非常好，那么它可能在今天，甚至是未来，都会有非常不错的收益。而tick数据，则能为你构建这种真实的历史环境，用于验证你的交易策略。

Tick数据还被用于量化交易的风险管理。详细的tick数据通过提供对市场流动性、滑点和订单执行质量等信息，来帮助交易员管理风险。Tick数据所提供的碎片化信息是市场微观结构研究员的重要工具。另外，tick数据甚至被用在法律监管行业，金融机构通常需要访问tick数据以满足法规合规和报告的要求。

区分高频数据的质量

与大部分产品一样，tick数据也有质量高低之分，其质量直接影响着后续的应用效果。低质量的tick数据主要表现在数据损坏上，下面是几种典型的数据损坏表现：

数据中断
由于网络中断或系统故障，ticks可能在某段时间内没有被记录下来，导致数据中断。
无效交易
指交易价格为零，甚至是负数的tick记录。比如在数据中发现USDEUR的价格是零，显然是错误的，这种数据需要移除。
重复数据
比如USDEUR在同一时刻出现了多个价格相同的ticks，这可能是因为数据记录不完整或系统错误导致的。
重复时间戳
比如某个货币对在某一时刻实际上没有变化，但由于系统错误添加了多个时间戳，使数据看起来变化非常频繁。这种情况也有可能是人为的。有些数据提供商为了让他们的数据看起来更新非常快，故意给旧的数据添加时间戳，导致ticks不断增加，而价格实际上没有任何变化。
排除人为因素，tick数据损坏的主要原因是由于数据量过大导致的。比如在外汇交易中，一个货币对一天可能有几千笔交易，将所有货币一天的交易都记录下来，数据量会非常庞大，一两周下来可能会积累几百万条数据。此外，信号质量低、信号丢失或信号延迟也可能导致高频数据序列的损坏。
想要清理或检查其中的错误是非常困难的。处理这种量级的数据，Excel会比较吃力，一般的数据提供商都会使用更专业的工具来清理，比如谷歌开发的OpenRefine，或者使用Python的Pandas来编写数据清理脚本。
不同市场的数据差异
外汇市场和股票市场的数据相差比较大。外汇市场属于场外交易，数据难以获取，而股票、商品这种中心化的市场，所有交易都发生在交易所内，他们会负责记录每一笔的交易变化。这种市场的差异也能影响数据的质量。

如何查询tick数据

下面演示一下如何通过Alltick的实时行情接口查询股票tick数据。

#1 获取K线

import time
import requests
import json

# Extra headers
test_headers = {
    'Content-Type':'application/json'
}

'''
github：
申请免费token：
官网：
将如下JSON进行url的encode，复制到http的查询字符串的query字段里
{"trace":"python_http_test1","data":{"code":"AAPL.US","kline_type":1,"kline_timestamp_end":0,"query_kline_num":2,"adjust_type":0}}
'''
test_url1 = '%7B%22trace%22%3A%22python_http_test1%22%2C%22data%22%3A%7B%22code%22%3A%22AAPL.US%22%2C%22kline_type%22%3A1%2C%22kline_timestamp_end%22%3A0%2C%22query_kline_num%22%3A2%2C%22adjust_type%22%3A0%7D%7D'

resp1 = requests.get(url=test_url1, headers=test_headers)

# Decoded text returned by the request
text1 = resp1.text
print(text1)

#2 获取成交报价

import time
import requests
import json

# Extra headers
test_headers = {
    'Content-Type':'application/json'
}

'''
github：
申请免费token：
官网：
将如下JSON进行url的encode，复制到http的查询字符串的query字段里
{"trace":"python_http_test2","data":{"symbol_list":[{"code": "700.HK"},{"code": "UNH.US"},{"code": "600416.SH"}]}}
'''
test_url1 = '%7B%22trace%22%3A%22python_http_test2%22%2C%22data%22%3A%7B%22symbol_list%22%3A%5B%7B%22code%22%3A%20%22700.HK%22%7D%2C%7B%22code%22%3A%20%22UNH.US%22%7D%2C%7B%22code%22%3A%20%22600416.SH%22%7D%5D%7D%7D'

resp1 = requests.get(url=test_url1, headers=test_headers)

# Decoded text returned by the request
text1 = resp1.text
print(text1)

3 通过websocket订阅获取实时股票行情数据

import json
import websocket    # pip install websocket-client

'''
github：
申请免费token：
官网：
'''

class Feed(object):

    def __init__(self):
        self.url = 'wss://quote.tradeswitcher.com/quote-stock-b-ws-api?token=e945d7d9-9e6e-4721-922a-7251a9d311d0-1678159756806' # 这里输入websocket的url
        self.ws = None

    def on_open(self, ws):
        """
        Callback object which is called at opening websocket.
        1 argument:
        @ ws: the WebSocketApp object
        """
        print('A new WebSocketApp is opened!')

        # 开始订阅（举个例子）
        sub_param = {
            "cmd_id": 22002,
            "seq_id": 123,
            "trace":"3baaa938-f92c-4a74-a228-fd49d5e2f8bc-1678419657806",
            "data":{
                "symbol_list":[
                    {
                        "code": "700.HK",
                        "depth_level": 5,
                    },
                    {
                        "code": "UNH.US",
                        "depth_level": 5,
                    },
                    {
                        "code": "600416.SH",
                        "depth_level": 5,
                    }
                ]
            }
        }

        #如果希望长时间运行，除了需要发送订阅之外，还需要修改代码，定时发送心跳，避免连接断开，具体查看接口文档
        sub_str = json.dumps(sub_param)
        ws.send(sub_str)
        print("depth quote are subscribed!")

    def on_data(self, ws, string, type, continue_flag):
        """
        4 argument.
        The 1st argument is this class object.
        The 2nd argument is utf-8 string which we get from the server.
        The 3rd argument is data type. ABNF.OPCODE_TEXT or ABNF.OPCODE_BINARY will be came.
        The 4th argument is continue flag. If 0, the data continue
        """

    def on_message(self, ws, message):
        """
        Callback object which is called when received data.
        2 arguments:
        @ ws: the WebSocketApp object
        @ message: utf-8 data received from the server
        """
        # 对收到的message进行解析
        result = eval(message)
        print(result)

    def on_error(self, ws, error):
        """
        Callback object which is called when got an error.
        2 arguments:
        @ ws: the WebSocketApp object
        @ error: exception object
        """
        print(error)

    def on_close(self, ws, close_status_code, close_msg):
        """
        Callback object which is called when the connection is closed.
        2 arguments:
        @ ws: the WebSocketApp object
        @ close_status_code
        @ close_msg
        """
        print('The connection is closed!')

    def start(self):
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_data=self.on_data,
            on_error=self.on_error,
            on_close=self.on_close,
        )
        self.ws.run_forever()
if __name__ == "__main__":
    feed = Feed()
    feed.start()

#4 获取{BANNED}最佳新盘口报价数据

import time
import requests
import json

# Extra headers
test_headers = {
    'Content-Type':'application/json'
}

'''
github：
申请免费token：
官网：
将如下JSON进行url的encode，复制到http的查询字符串的query字段里
{"trace":"python_http_test2","data":{"symbol_list":[{"code": "700.HK"},{"code": "UNH.US"},{"code": "600416.SH"}]}}
'''
test_url1 = '%7B%22trace%22%3A%22python_http_test2%22%2C%22data%22%3A%7B%22symbol_list%22%3A%5B%7B%22code%22%3A%20%22700.HK%22%7D%2C%7B%22code%22%3A%20%22UNH.US%22%7D%2C%7B%22code%22%3A%20%22600416.SH%22%7D%5D%7D%7D'

resp1 = requests.get(url=test_url1, headers=test_headers)

# Decoded text returned by the request
text1 = resp1.text
print(text1)

AllTick如何保证Tick数据质量？

作为专业的高频数据供应商，AllTick深知确保数据的准确性、可靠性和质量至关重要。我们在提供tick数据时始终采用以客户为中心的方法。我们深知tick数据准确性对于您的分析和盈利交易决策的重要性。

一些数据提供者可能会反复转发相同的时间。同样地，有些可能只提供多条记录中的一个价格，并通过生成自己的时间戳来记录。这两种方式都会导致数据质量较差，长期来看可能会对您的分析和交易产生不利影响。

我们从多个来源捕获tick数据，而不是依赖于一个地方。我们的数据涵盖每一次价格报价的情况。我们实时将高频数据发送给客户。我们所有的数据传输都采用SSL加密。因此，您可以信赖我们提供经过验证、准确和可靠的tick数据。

如何从AllTick获取高频数据API接口？

我们的数据涵盖外汇、港股CFD、美股CFD、商品和加密货币等领域的行情数据接口。这些接口专为交易所、开发者、量化团队、金融科技公司和专业机构设计。它们适用于所有主要客户端的开发语言，并提供完整的接口文档。我们也提供免费试用，方便做各种接入测试。