Storage API¶
High-level functions for persisting OHLCV time series data to ArcticDB.
Overview¶
The chronos_lab.storage module provides functions for storing time series data in ArcticDB, a high-performance time series database with versioning support.
Functions¶
chronos_lab.storage.ohlcv_to_arcticdb ¶
ohlcv_to_arcticdb(*, ohlcv: DataFrame | Dict[str, DataFrame], library_name: Optional[str] = None, adb_mode: str = 'write') -> Dict[str, int]
Store OHLCV data to ArcticDB library for persistent time series storage.
Accepts OHLCV data in multiple formats and stores it in ArcticDB with symbol-level organization. Automatically splits MultiIndex DataFrames by symbol for efficient per-symbol versioning and retrieval.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ohlcv
|
DataFrame | Dict[str, DataFrame]
|
OHLCV data in one of two formats: - MultiIndex DataFrame with ('date', 'id'/'symbol') levels - Dictionary mapping symbols to individual DataFrames |
required |
library_name
|
Optional[str]
|
ArcticDB library name for storage. If None, uses ARCTICDB_DEFAULT_LIBRARY_NAME from ~/.chronos_lab/.env configuration. |
None
|
adb_mode
|
str
|
Storage mode for ArcticDB operations: - 'write': Overwrite existing data (default) - 'append': Append new data to existing symbols |
'write'
|
Returns:
| Type | Description |
|---|---|
Dict[str, int]
|
Dictionary with status information: - 'statusCode': 0 on success, -1 on failure, 1 if some symbols failed - 'skipped_symbols': List of symbols that failed to store (if any) |
Raises:
| Type | Description |
|---|---|
None
|
Errors are logged but not raised. Check statusCode in return value. |
Examples:
Store MultiIndex DataFrame from Yahoo Finance: >>> from chronos_lab.sources import ohlcv_from_yfinance >>> from chronos_lab.storage import ohlcv_to_arcticdb >>> >>> # Fetch data >>> prices = ohlcv_from_yfinance( ... symbols=['AAPL', 'MSFT', 'GOOGL'], ... period='1y' ... ) >>> >>> # Store in ArcticDB >>> result = ohlcv_to_arcticdb( ... ohlcv=prices, ... library_name='yfinance', ... adb_mode='write' ... ) >>> if result['statusCode'] == 0: ... print("Successfully stored data")
Store dictionary of DataFrames from Intrinio: >>> from chronos_lab.sources import ohlcv_from_intrinio >>> >>> # Fetch as dictionary >>> prices_dict = ohlcv_from_intrinio( ... symbols=['AAPL', 'MSFT'], ... period='3mo', ... interval='daily', ... output_dict=True ... ) >>> >>> # Store dictionary directly >>> result = ohlcv_to_arcticdb( ... ohlcv=prices_dict, ... library_name='intrinio' ... )
Append new data to existing symbols: >>> # Fetch latest data >>> new_prices = ohlcv_from_yfinance( ... symbols=['AAPL', 'MSFT'], ... period='1d' ... ) >>> >>> # Append to existing data >>> result = ohlcv_to_arcticdb( ... ohlcv=new_prices, ... library_name='yfinance', ... adb_mode='append' ... )
Handle partial failures: >>> result = ohlcv_to_arcticdb( ... ohlcv=prices, ... library_name='yfinance' ... ) >>> if result['statusCode'] == 1: ... print(f"Failed symbols: {result['skipped_symbols']}") >>> elif result['statusCode'] == 0: ... print("All symbols stored successfully")
Note
- Input DataFrame must have exactly 2-level MultiIndex: ('date', 'id'/'symbol')
- Each symbol is stored as a separate versioned entity in ArcticDB
- Storage mode 'write' overwrites existing data; use 'append' to add new rows
- Previous versions are pruned automatically to save space
- All timestamps should be UTC timezone-aware