是否有任何优雅的方法来定义具有 dtype 数组列的数据框？

Question

我想在 pandas 中处理 stock level-2 数据。为简单起见，假设每行有四种数据：

• 毫秒：时间戳，int64
• last_price：最后交易价格，float64，
• ask_queue：ask端的体积，一个固定大小（200）的int32数组
• bid_queue：bid_queue的数量，一个固定大小（200）的int32数组

在 numpy 中可以很容易地定义为结构化 dtype：

dtype = np.dtype([
   ('millis', 'int64'), 
   ('last_price', 'float64'), 
   ('ask_queue', ('int32', 200)), 
   ('bid_queue', ('int32', 200))
])

这样，我就可以访问ask_queue 和bid_queue 之类的：

In [17]: data = np.random.randint(0, 100, 1616 * 5).view(dtype)

% compute the average of ask_queue level 5 ~ 10
In [18]: data['ask_queue'][:, 5:10].mean(axis=1)  
Out[18]: 
array([33.2, 51. , 54.6, 53.4, 15. , 37.8, 29.6, 58.6, 32.2, 51.6, 34.4,
       43.2, 58.4, 26.8, 54. , 59.4, 58.8, 38.8, 35.2, 71.2])

我的问题是如何定义一个DataFrame包含的数据？

这里有两种解决方案：

A.将ask_queue 和bid_queue 设置为两列，数组值如下：

In [5]: df = pd.DataFrame(data.tolist(), columns=data.dtype.names)

In [6]: df.dtypes
Out[6]: 
millis          int64
last_price    float64
ask_queue      object
bid_queue      object
dtype: object

但是，这个解决方案至少存在两个问题：

1. ask_queue 和 bid_queue 丢失了 2D 数组的 dtype 和所有 方便的方法；
2. 性能，因为它变成了对象数组而不是 2D 数组。

B.将ask_queue 和bid_quene 展平为2 * 200 列：

In [8]: ntype = np.dtype([('millis', 'int64'), ('last_price', 'float64')] + 
   ...:                  [(f'{name}{i}', 'int32') for name in ['ask', 'bid'] for i in range(200)])

In [9]: df = pd.DataFrame.from_records(data.view(ntype))

In [10]: df.dtypes
Out[10]: 
millis          int64
last_price    float64
ask0            int32
ask1            int32
ask2            int32
ask3            int32
ask4            int32
ask5            int32
...

比解决方案 A 好。但是 2 * 200 列看起来是多余的。

是否有任何解决方案可以利用 numpy 中的结构化 dtype 的优势？ 我想知道ExtensionArray 或 `ExtensionDtype' 是否可以解决这个问题。

Answer 1

问：是否有任何解决方案可以利用 numpy 中的结构化 dtype 的优势？

与仅使用 ToB（Top-of-the-Book）价格馈送数据相比，使用 L2-DoM 数据具有双重复杂性。 a) 本机提要速度很快（非常快/FIX 协议或其他私有数据提要）每毫秒提供数百、数千（在专业的基本事件期间更多）L2-DoM 变化的记录。处理和存储必须以性能为导向 b) 由于 a) 项的性质，任何类型的离线分析都必须成功操作和有效处理大型数据集

• 存储首选项
• 使用numpy-类似语法首选项
• 性能偏好

---

存储偏好：已解决

鉴于 pandas.DataFrame 被设置为首选存储类型，让我们尊重这一点，即使语法和性能偏好可能会产生不利影响。

采用其他方式是可能的，但可能会引入未知的重构/重新设计成本，而 O/P 的运营环境不需要或已经不愿意承担这些成本。

话虽如此，pandas 功能限制必须纳入设计考虑因素，所有其他步骤都必须遵守，除非此偏好可能会在未来某个时间得到修改。

---

numpy-类似的语法：已解决

这个要求是合理而清晰的，因为numpy 工具是为高性能数字运算而设计的快速而智能的工具。鉴于设置的存储偏好，我们将在.STORE 和@987654332 上以合理的成本实现一对numpy-tricks 以适应pandas 2D-DataFrame @方向：

 # on .STORE:
 testDF['ask_DoM'][aRowIDX] = ask200.dumps()      # type(ask200) <class 'numpy.ndarray'>

 # on .RETRIEVE:
 L2_ASK = np.loads( testDF['ask_DoM'][aRowIDX] )  # type(L2_ASK) <class 'numpy.ndarray'>

---

性能偏好：测试

针对.STORE 和.RETRIEVE 方向的建议解决方案的净附加成本经过测试：

一次性费用在.STORE 方向上不少于70 [us] 且不超过~ 160 [us] 每个单元格对于给定比例的 L2_DoM 数组（平均：78 [ms] StDev：9-11 [ms]）：

>>> [ f( [testDUMPs() for _ in range(1000)] ) for f in (np.min,np.mean,np.std,np.max) ]
[72, 79.284, 11.004153942943548, 150]
[72, 78.048, 10.546135548152224, 160]
[71, 78.584,  9.887971227708949, 139]
[72, 76.9,    8.827332496286745, 132]

在.RETRIEVE 方向上的重复成本 不小于 46 [us] 且不超过 ~ 123 [us] 每个单元格对于给定L2_DoM 数组的尺度（平均：50 [us] StDev：9.5 [us]）：

>>> [ f( [testLOADs() for _ in range(1000)] ) for f in (np.min,np.mean,np.std,np.max) ]
[46, 50.337, 9.655194197943405, 104]
[46, 49.649, 9.462272665697178, 123]
[46, 49.513, 9.504293766503643, 123]
[46, 49.77,  8.367165350344164, 114]
[46, 51.355, 6.162434583831296,  89]

如果使用更好的架构对齐int64 数据类型（是的，以存储成本翻倍为代价，但计算成本将决定这一举措是否具有性能优势）和从有机会使用基于 memoryview 的操作，这可以减少喉咙并将附加延迟减少到大约 22 [us]。

---

_{测试在 py3.5.6、numpy v1.15.2 下运行，使用：}

>>> import numpy as np; ask200 = np.arange( 200, dtype = np.int32 ); s = ask200.dumps()
>>> from zmq import Stopwatch; aClk = Stopwatch()
>>> def testDUMPs():
...     aClk.start()
...     s = ask200.dumps()
...     return aClk.stop()
... 
>>> def testLOADs():
...     aClk.start()
...     a = np.loads( s )
...     return aClk.stop()
...

_{平台 CPU、缓存层次结构和 RAM 详细信息：}

>>> get_numexpr_cpuinfo_details_on_CPU()

'TLB size'______________________________:'1536 4K pages'
'address sizes'_________________________:'48 bits physical, 48 bits virtual'
'apicid'________________________________:'17'
'bogomips'______________________________:'7199.92'
'bugs'__________________________________:'fxsave_leak sysret_ss_attrs null_seg spectre_v1 spectre_v2'
'cache size'____________________________:'2048 KB'
'cache_alignment'_______________________:'64'
'clflush size'__________________________:'64'
'core id'_______________________________:'1'
'cpu MHz'_______________________________:'1400.000'
'cpu cores'_____________________________:'2'
'cpu family'____________________________:'21'
'cpuid level'___________________________:'13'
'flags'_________________________________:'fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush mmx fxsr sse sse2 ht syscall nx mmxext fxsr_opt pdpe1gb rdtscp lm constant_tsc rep_good nopl nonstop_tsc extd_apicid aperfmperf eagerfpu pni pclmulqdq monitor ssse3 cx16 sse4_1 sse4_2 popcnt aes xsave avx lahf_lm cmp_legacy svm extapic cr8_legacy abm sse4a misalignsse 3dnowprefetch osvw ibs xop skinit wdt lwp fma4 nodeid_msr topoext perfctr_core perfctr_nb cpb hw_pstate vmmcall arat npt lbrv svm_lock nrip_save tsc_scale vmcb_clean flushbyasid decodeassists pausefilter pfthreshold'
'fpu'___________________________________:'yes'
'fpu_exception'_________________________:'yes'
'initial apicid'________________________:'1'
'microcode'_____________________________:'0x6000626'
'model'_________________________________:'1'
'model name'____________________________:'AMD FX(tm)-4100 Quad-Core Processor'
'physical id'___________________________:'0'
'power management'______________________:'ts ttp tm 100mhzsteps hwpstate cpb'
'processor'_____________________________:'1'
'siblings'______________________________:'4'
'stepping'______________________________:'2'
'vendor_id'_____________________________:'AuthenticAMD'
'wp'____________________________________:'yes'

Answer 2

Pandas 旨在处理和处理二维数据（您可以放入电子表格的那种数据）。因为“ask_queue”和“bid_queue”不是一维序列而是二维数组，所以不能（轻松）将它们推送到 Pandas 数据帧中。

在这种情况下，您必须使用其他库，例如 xarray：http://xarray.pydata.org/

import xarray as xr

# Creating variables, first argument is the name of the dimensions
last_price = xr.Variable("millis", data["last_price"])
ask_queue = xr.Variable(("millis", "levels"), data["ask_queue"])
bid_queue = xr.Variable(("millis", "levels"), data["bid_queue"])

# Putting the variables in a dataset, the multidimensional equivalent of a Pandas
# dataframe
ds = xr.Dataset({"last_price": last_price, "ask_queue": ask_queue,
                 "bid_queue": bid_queue}, coords={"millis": data["millis"]})

# Computing the average of ask_queue level 5~10
ds["ask_queue"][{"levels": slice(5,10)}].mean(axis=1)