# 数据流水线 (/zh/features/data/financial-data-pipelines)

如何将原始下载转化为可供分析且保持最新的数据？VBT 将所有标的和特征保存在同一 `Data` 对象中，对齐不同历史、统一时区、增量更新并直接运行指标，让 Python
市场数据流水线只需几次方法调用。

```python title="Download 50 stocks, update them a month later, and compute RSI on all of them"
>>> symbols = [
...     "AAPL", "MSFT", "AMZN", "GOOGL", "META", "NVDA", "JPM", "V", "MA", "UNH",
...     "HD", "PG", "JNJ", "XOM", "CVX", "KO", "PEP", "MRK", "ABBV", "LLY",
...     "COST", "WMT", "BAC", "WFC", "CSCO", "ORCL", "INTC", "AMD", "QCOM", "TXN",
...     "ADBE", "CRM", "NFLX", "DIS", "NKE", "MCD", "SBUX", "BA", "CAT", "DE",
...     "GE", "HON", "IBM", "MMM", "UPS", "LOW", "T", "VZ", "PFE", "TMO",
... ]
>>> data = vbt.YFData.pull(symbols, start="2024-01-01", end="2024-06-01")
>>> data.close.shape, data.index[-1].date()
((105, 50), datetime.date(2024, 5, 31))

>>> data = data.update(end="2024-07-01")  # (1)
>>> data.close.shape, data.index[-1].date()
((124, 50), datetime.date(2024, 6, 28))

>>> rsi = data.run("rsi", window=14).rsi  # (2)
>>> rsi.iloc[-1].sort_values(ascending=False).head(5).round(1)
rsi_window  symbol
14          LLY       75.8
            T         75.4
            ADBE      74.5
            ORCL      66.5
            HON       63.6
Name: 2024-06-28 00:00:00-04:00, dtype: float64
```

1.  为每个标的刷新最后一根已存 K 线，并追加之后的 K 线。
2.  一次对所有标的运行指标，自动选取正确的输入特征。

## 数据容器 \[#data-containers]

`Data` 对象将每个标的映射到开盘、最高、最低、收盘和成交量等特征。特征可按名称访问，例如 `data.close` 或 `data.get("Volume")`；`data.hlc3`
等派生价格则按需计算。每个特征返回一个每标的一列的
DataFrame，可直接用于指标和回测。同一容器也可按特征组织，适合以每字段一张表形式到来的数据，如下方[面向特征的数据](#feature-oriented-data)功能介绍所示。无需离开对象，即可选择、重命名、添加和删除标的及特征。

通过 `vbt.Data.from_data`，您自己的数据可以成为数据对象。它接受每个标的一张 DataFrame，因此许多导出和数据库产生的带标的列长表，需要先拆分：

```python title="Turn a long table of daily bars into a data object"
>>> long = pd.DataFrame({
...     "date": pd.to_datetime(["2024-01-02", "2024-01-03", "2024-01-02", "2024-01-03"], utc=True),
...     "symbol": ["AAPL", "AAPL", "MSFT", "MSFT"],
...     "close": [185.64, 184.25, 370.87, 370.60],
...     "volume": [82488700, 58414500, 25258600, 23083500],
... })
>>> data = vbt.Data.from_data({
...     symbol: df.set_index("date").drop(columns="symbol")
...     for symbol, df in long.groupby("symbol")
... })
>>> data.symbols, data.features
(['AAPL', 'MSFT'], ['close', 'volume'])

>>> data.close
symbol                       AAPL    MSFT
date
2024-01-02 00:00:00+00:00  185.64  370.87
2024-01-03 00:00:00+00:00  184.25  370.60
```

设置 `columns_are_symbols=True` 后，也可使用由宽格式 DataFrame
组成的字典，每个特征一张表、标的作为列。这适合已按独立表格提供的价格、基本面、模型评分或投资组合时间序列。自定义提供商还可实现 `fetch_feature` 和
`update_feature`，以跨多个标的获取及更新单个特征。

## 对齐 \[#alignment]

不同标的很少拥有完全相同的历史。一起拉取时，会合并它们的索引，并用 `NaN` 填充缺失 K 线；也可设置 `missing_index="drop"`
删除缺失位置。无论哪种方式，索引不匹配时都会发出警告：

```python title="Align a symbol that started trading later"
>>> data = vbt.YFData.pull(["BTC-USD", "SOL-USD"], start="2020-01-01", end="2021-01-01")
>>> data.close.shape
(366, 2)
>>> data.close.apply(lambda x: x.first_valid_index().date()).tolist()
[datetime.date(2020, 1, 1), datetime.date(2020, 4, 10)]

>>> dropped = vbt.YFData.pull(
...     ["BTC-USD", "SOL-USD"],
...     start="2020-01-01",
...     end="2021-01-01",
...     missing_index="drop",
... )
>>> dropped.close.shape, dropped.index[0].date()
((266, 2), datetime.date(2020, 4, 10))
```

时区也以同样方式处理：加载时可转换为同一时区，返回不同时区的标的会被报告，而非静默错位。加载时索引会排序，重复时间戳保留最后一个值。

研究各资产完整历史时，可保留缺失观测；比较资产在相同日期的表现时，则使用共同区间。对齐不会为缺失 K
线凭空创造价格。对于在不同时间可获知的日频和日内输入，请通过[多时间周期分析](/zh/features/data/multi-timeframe-analysis/)按可用时间对齐。

## 更新 \[#updates]

`update` 只获取新增内容。传入 `return_meta=True` 可查看追加了哪些 K 线，以及提供商修订了哪些重叠 K 线；如果上次更新时最后一根 K
线仍在形成，这一点很重要。固定计划可使用 `vbt.DataUpdater`
按间隔更新，并在每次更新后调用您的代码，这是实时流水线的基础，详见[实时模拟](/zh/features/backtesting/live-simulation/#live-pipelines)页面。

请对返回对象赋值，例如
`data = data.update()`。保留并延伸原始历史，能让重复指标计算拥有相同起点。对于指数均线等计算，这一点很重要；如果每次都从不同的最近窗口开始，数值可能发生变化。

如果只需要更新区间，使用
`concat=False`。这适合通过[数据保存器](/zh/features/data/data-storage-and-databases/#keeping-stores-current)将每个批次写入磁盘的采集程序。需要在观测间保留指标状态的计算，请参阅[流式指标](/zh/features/indicators/streaming-indicators/)。

已经通过外部行情流接收 K 线？您的集成可以将每批数据封装为数据对象，与先前数据合并，也可提供自己的更新方法。合并时，后到数据替换重叠观测，因此修订 K 线可替换旧版本。行情连接和遗漏 K
线的恢复仍由您的集成负责。

## 变换与合并 \[#transform-and-merge]

`transform` 将数据作为一个 DataFrame 传给任意函数，并返回新数据对象，涵盖清洗、筛选和添加特征。`merge`
组合数据对象，例如来自两个提供商的对象；拼接则连接连续时段。原始文件和 DataFrame 通过解析成为数据对象，各处都能识别 `"15 minutes"`、`"1h"`
等时间周期。下方[数据解析](#data-parsing)、[数据变换](#data-transformation)和[数据合并](#data-merging)功能介绍逐步展示这些操作。

## 在数据上运行函数 \[#running-functions-on-data]

`data.run()` 接受指标名称、函数或表达式，并按名称填充所需特征，因此 `data.run("rsi")` 读取收盘价，`data.run("atr")`
读取最高、最低和收盘价。不同函数可使用不同参数，一次调用可运行多个函数，需要时还可并行执行，如[可运行数据](#runnable-data)和[并行数据](#parallel-data)功能介绍所示。

## 将研究特征与价格保存在一起 \[#keep-research-features-beside-prices]

您可将计算特征附加到同一数据对象中，让价格、指标输出和模型评分一起进入下一研究阶段。计算可以是普通 Python 函数，其参数名称与数据字段匹配：

```python title="Keep a custom feature beside prices"
>>> research_close = pd.DataFrame(
...     {"AAA": [100.0, 102.0, 104.0], "BBB": [50.0, 49.0, 51.0]},
...     index=pd.date_range("2024-01-01", periods=3, freq="D", tz="UTC"),
... )
>>> research_data = vbt.Data.from_data(
...     vbt.feature_dict(
...         {
...             "Close": research_close,
...             "High": research_close + 2,
...             "Low": research_close - 2,
...         }
...     )
... )

>>> def range_pct(high, low, close):
...     return 100 * (high - low) / close

>>> range_values = research_data.run(range_pct)
>>> research_data = research_data.add_feature("RangePct", range_values)
>>> research_data.features
['Close', 'High', 'Low', 'RangePct']

>>> research_data.get("RangePct").round(2)
                            AAA   BBB
2024-01-01 00:00:00+00:00  4.00  8.00
2024-01-02 00:00:00+00:00  3.92  8.16
2024-01-03 00:00:00+00:00  3.85  7.84
```

一次计算便为两个示例标的同时添加最高价与最低价之差占收盘价百分比的特征。之后可以按名称选择，或使用[存储方法](/zh/features/data/data-storage-and-databases/)与其他特征一起保存。

## 标的池与期货 \[#universes-and-futures]

标的可携带行业或交易所等类别，成为用于分组的列层级。研究指数成分股时，请构建记录各标的在每个日期是否属于指数成分股的成员表，并作为掩码使用，避免研究范围包含后来才加入的股票。连续期货序列应在加载前完成换月和向后调整，避免将换月跳空误认为价格变动。

## 错误 \[#errors]

使用 `skip_on_error=True` 让大规模下载继续进行：VBT 收集可用标的，并报告失败标的。稍后重试失败标的，再合并到数据集。

## 面向特征的数据 \[#feature-oriented-data]

新增于 1.14.0

✅ 按使用方式组织数据。VBT 同时支持面向标的和面向特征的存储：每个交易品种一张表，或每个开盘、最高、最低、收盘等字段一张表。面向特征的数据让这些价格表可直接跨标的使用，随时用于指标和回测。

```python title="Create a feature-oriented data instance from various portfolio time series"
>>> data = vbt.YFData.pull(["AAPL", "MSFT", "GOOG"], tz_convert="UTC")
>>> pf = data.run("from_random_signals", n=[10, 20, 30], seed=42)

>>> pf_data = vbt.Data.from_data(
...     vbt.feature_dict({
...         "cash": pf.cash,
...         "assets": pf.assets,
...         "asset_value": pf.asset_value,
...         "value": pf.value
...     })
... )
>>> pf_data.get(feature="cash", symbol=(10, "AAPL"))
Date
1980-12-12 05:00:00+00:00      100.000000
1980-12-15 05:00:00+00:00      100.000000
1980-12-16 05:00:00+00:00      100.000000
1980-12-17 05:00:00+00:00      100.000000
1980-12-18 05:00:00+00:00      100.000000
                                      ...
2023-08-25 04:00:00+00:00     1837.648268
2023-08-28 04:00:00+00:00     1837.648268
2023-08-29 04:00:00+00:00     1837.648268
2023-08-30 04:00:00+00:00     1837.648268
2023-08-31 04:00:00+00:00     1837.648268
Name: (10, AAPL), Length: 10770, dtype: float64
```

## 并行数据 \[#parallel-data]

新增于 1.10.0

✅ 数据获取和更新可以轻松并行执行。

```python title="Benchmark fetching multiple symbols serially and concurrently"
>>> symbols = ["SPY", "TLT", "XLF", "XLE", "XLU", "XLK", "XLB", "XLP", "XLY", "XLI", "XLV"]

>>> with vbt.Timer() as timer:
...     data = vbt.YFData.pull(symbols)
>>> print(timer.elapsed())
4.52 seconds

>>> with vbt.Timer() as timer:
...     data = vbt.YFData.pull(symbols, execute_kwargs=dict(engine="threadpool"))
>>> print(timer.elapsed())
918.54 milliseconds
```

## 可运行数据 \[#runnable-data]

新增于 1.5.0

✅
厌倦了弄清指标需要哪些参数？数据实例现在可识别任意指标或函数的参数，将其映射到列名，并传入所需列来运行函数。您还可修改映射、覆盖指标参数，以及按名称查询指标。数据实例将搜索所有集成指标包，返回找到的第一个（也是最佳）匹配。

```python title="Run Stochastic RSI by data"
>>> data = vbt.YFData.pull("BTC-USD")
>>> stochrsi = data.run("stochrsi")
>>> stochrsi.fastd
Date
2014-09-17 00:00:00+00:00          NaN
2014-09-18 00:00:00+00:00          NaN
2014-09-19 00:00:00+00:00          NaN
2014-09-20 00:00:00+00:00          NaN
2014-09-21 00:00:00+00:00          NaN
                                   ...
2023-01-15 00:00:00+00:00    96.168788
2023-01-16 00:00:00+00:00    91.733393
2023-01-17 00:00:00+00:00    78.295255
2023-01-18 00:00:00+00:00    48.793133
2023-01-20 00:00:00+00:00    26.242474
Name: Close, Length: 3047, dtype: float64
```

## 数据解析 \[#data-parsing]

新增于 1.3.0

✅ 厌倦了将开盘、最高、最低、收盘价作为独立时间序列传入？投资组合类方法现在不仅接受收盘价，还可接受数据实例并自动提取其中 OHLC 数据。这个小巧实用的功能为您节省时间！

```python title="Backtest random signals from a data instance"
>>> data = vbt.YFData.pull("BTC-USD", start="2020-01", end="2020-03")
>>> pf = vbt.PF.from_random_signals(data, n=10, seed=42)
```

## 数据变换 \[#data-transformation]

新增于 1.2.3

✅ 获取数据后如何修改？新增方法将所有标的放入单一 DataFrame，并传给用户定义函数进行变换。

```python title="Remove weekends"
>>> data = vbt.YFData.pull(["BTC-USD", "ETH-USD"], start="2020-01-01", end="2020-01-14")
>>> new_data = data.transform(lambda df: df[~df.index.weekday.isin([5, 6])])
>>> new_data.close
symbol                         BTC-USD     ETH-USD
Date
2020-01-01 00:00:00+00:00  7200.174316  130.802002
2020-01-02 00:00:00+00:00  6985.470215  127.410179
2020-01-03 00:00:00+00:00  7344.884277  134.171707
2020-01-06 00:00:00+00:00  7769.219238  144.304153
2020-01-07 00:00:00+00:00  8163.692383  143.543991
2020-01-08 00:00:00+00:00  8079.862793  141.258133
2020-01-09 00:00:00+00:00  7879.071289  138.979202
2020-01-10 00:00:00+00:00  8166.554199  143.963776
2020-01-13 00:00:00+00:00  8144.194336  144.226593
```

## 数据合并 \[#data-merging]

新增于 1.1.0

✅ 经常需要将不同交易所的标的放入同一篮子回测。为此，VBT 提供将多个数据实例合并为单个实例的类方法。不仅能组合多个标的，也能合并同一标的的数据集，全程自动完成！

```python title="Pull BTC datasets from various exchanges and plot the last full year relative to their mean"
>>> binance_data = vbt.CCXTData.pull("BTCUSDT", exchange="binance")
>>> bybit_data = vbt.CCXTData.pull("BTCUSDT", exchange="bybit")
>>> bitfinex_data = vbt.CCXTData.pull("BTC/USDT", exchange="bitfinex")
>>> kucoin_data = vbt.CCXTData.pull("BTC-USDT", exchange="kucoin")

>>> data = vbt.Data.merge([
...     binance_data.rename({"BTCUSDT": "Binance"}),
...     bybit_data.rename({"BTCUSDT": "Bybit"}),
...     bitfinex_data.rename({"BTC/USDT": "Bitfinex"}),
...     kucoin_data.rename({"BTC-USDT": "KuCoin"}),
... ], missing_index="drop", silence_warnings=True)

>>> @njit
... def rescale_nb(x):
...     return (x - x.mean()) / x.mean()

>>> rescaled_close = data.close.vbt.row_apply(rescale_nb)
>>> rescaled_close = rescaled_close.vbt.rolling_mean(30)
>>> last_full_year = rescaled_close.index[-1].year - 1
>>> rescaled_close.loc[str(last_full_year)].vbt.plot().show()
```

最近完整日历年中，Binance、Bybit、Bitfinex 和 KuCoin 的重新缩放 BTC 价格. [Figure data (JSON)](/assets/figures/features/data/data-merging.ecc16eb2ecbd.json)