Storage API¶

High-level functions for persisting time series data and structured datasets.

Overview¶

The chronos_lab.storage module provides functions for storing data:

Time Series: Store OHLCV data in ArcticDB (high-performance time series database)
Datasets: Store structured data (portfolios, watchlists, metadata) locally or in DynamoDB
Store: Store arbitrary files (images, JSON, binary data) to local filesystem and/or S3

Time Series Functions¶

chronos_lab.storage.ohlcv_to_arcticdb ¶

ohlcv_to_arcticdb(*, ohlcv: DataFrame | Dict[str, DataFrame], backend: Optional[str] = None, 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
`backend`	`Optional[str]`	Storage backend type ('s3', 'lmdb', or 'mem', case-insensitive). If None, uses ARCTICDB_DEFAULT_BACKEND from ~/.chronos_lab/.env configuration. Overrides the default backend setting for this operation.	`None`
`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 with default backend >>> result = ohlcv_to_arcticdb( ... ohlcv=prices, ... library_name='yfinance', ... adb_mode='write' ... ) >>> if result['statusCode'] == 0: ... print("Successfully stored data")

Store with explicit backend selection: >>> # Store to S3 backend explicitly >>> result = ohlcv_to_arcticdb( ... ohlcv=prices, ... backend='s3', ... library_name='yfinance', ... adb_mode='write' ... ) >>> >>> # Store to local LMDB backend explicitly >>> result = ohlcv_to_arcticdb( ... ohlcv=prices, ... backend='lmdb', ... library_name='yfinance' ... )

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

Dataset Functions¶

chronos_lab.storage.to_dataset ¶

to_dataset(dataset_name: str, dataset: Dict[str, Any]) -> Dict[str, int]

Save a dataset to local JSON file or DynamoDB table.

Stores structured dataset dictionary based on naming convention. Automatically creates parent directories for local storage if needed.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier. Use 'ddb_' prefix for DynamoDB datasets, no prefix for local JSON files.	required
`dataset`	`Dict[str, Any]`	Dictionary mapping keys to item attribute dictionaries	required

Returns:

Type	Description
`Dict[str, int]`	Dictionary with 'statusCode': 0 on success, -1 on failure

Examples:

Save to local JSON file: >>> from chronos_lab.storage import to_dataset >>> >>> data = { ... 'AAPL': {'name': 'Apple Inc.', 'sector': 'Technology', 'price': 175.50}, ... 'MSFT': {'name': 'Microsoft', 'sector': 'Technology', 'price': 420.00} ... } >>> result = to_dataset(dataset_name='tech_stocks', dataset=data) >>> if result['statusCode'] == 0: ... print("Dataset saved successfully")

Save to DynamoDB: >>> data = { ... 'security1': {'ticker': 'AAPL', 'exchange': 'NASDAQ'}, ... 'security2': {'ticker': 'MSFT', 'exchange': 'NASDAQ'} ... } >>> result = to_dataset(dataset_name='ddb_securities', dataset=data)

Note

Local datasets: Saved to {DATASET_LOCAL_PATH}/{name}.json
DynamoDB datasets: Require DATASET_DDB_TABLE_NAME and DATASET_DDB_MAP configuration
DynamoDB items automatically get 'pk' and 'sk' keys from dataset map
Datetime objects serialized as ISO strings
Local file parent directories created automatically

File Storage Functions¶

chronos_lab.storage.to_store ¶

to_store(*, file_name: str, content: bytes, folder: Optional[str] = None, stores: Optional[List[str]] = None, s3_metadata: Optional[Dict[str, str]] = None, s3_put_object_kwargs: Optional[Dict[str, Any]] = None)

Store file content to local filesystem and/or S3 based on configuration.

Saves arbitrary file content (images, JSON, binary data) to configured storage backends. Supports local filesystem and S3, or both simultaneously.

Parameters:

Name	Type	Description	Default
`file_name`	`str`	Name of the file to save.	required
`content`	`bytes`	File content as bytes.	required
`folder`	`Optional[str]`	Optional subdirectory/prefix within the configured storage path. For local storage, creates subdirectory under STORE_LOCAL_PATH. For S3, prepends to object key. Defaults to None.	`None`
`stores`	`Optional[List[str]]`	List of storage backends to use. Options: ['local'], ['s3'], or ['local', 's3'] for both. Defaults to ['local'].	`None`
`s3_metadata`	`Optional[Dict[str, str]]`	Optional metadata dict to attach to S3 object (ignored for local). Defaults to None.	`None`
`s3_put_object_kwargs`	`Optional[Dict[str, Any]]`	Additional arguments passed to boto3 put_object (e.g., ContentType, ACL). Defaults to None.	`None`

Returns:

Type	Description
	Dictionary with status information: - 'local_statusCode': 0 on success, -1 on failure (if 'local' in stores) - 'file_path': Full local path to saved file (if successful) - 's3_statusCode': 0 on success, -1 on failure (if 's3' in stores) - 's3_client_response': boto3 response dict (if S3 successful)

Example

from chronos_lab.plot import plot_ohlcv_anomalies
from chronos_lab.storage import to_store

# Generate a plot
plot_data = plot_ohlcv_anomalies(anomalies_df, plot_to_store=False)

# Save to both local and S3
result = to_store(
    file_name=plot_data['file_name'],
    content=plot_data['content'],
    folder='anomaly_charts',
    stores=['local', 's3'],
    s3_metadata={'symbol': 'AAPL', 'analysis_type': 'anomaly'}
)

if result['local_statusCode'] == 0:
    print(f"Saved locally to: {result['file_path']}")
if result['s3_statusCode'] == 0:
    print("Saved to S3 successfully")