Skip to content

data.data_repository

data.data_repository

DataRepository

Bases: Generic[TConfig]

Retrieves data from either internal storage, a broker, or both. Data that is not already in storage is automatically added.

get_data_and_store staticmethod 

get_data_and_store(broker: HistoricalBroker[TConfig], config: TConfig, expected_timestamps: Iterable[datetime], database_path: str | Path | None = None) -> list[MarketDataEvent]

Retrieves data from either the database, a broker, or both. Any data not in the database that is retrieved from the broker is then put into storage for further use.

Parameters:

Name Type Description Default
broker HistoricalBroker[TConfig]

The historical broker to derive data from if necessary.

 required
config TConfig

The corresponding config to the historical broker.

 required
expected_timestamps Iterable[datetime]

An iterable of the expected timestamps.

 required
database_path str | Path | None

Where the database should live. If not provided, DataStorage resolves it via the OS-standard user data directory.

 None
Source code in src\contango\data\data_repository.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@staticmethod
def get_data_and_store(
    broker: HistoricalBroker[TConfig],
    config: TConfig,
    expected_timestamps: Iterable[datetime],
    database_path: str | Path | None = None,
) -> list[MarketDataEvent]:
    """
    Retrieves data from either the database, a broker, or both.
    Any data not in the database that is retrieved from the broker is then put into storage for further use.

    Args:
        broker: The historical broker to derive data from if necessary.
        config: The corresponding config to the historical broker.
        expected_timestamps: An iterable of the expected timestamps.
        database_path: Where the database should live. If not provided,
                       DataStorage resolves it via  the OS-standard
                       user data directory.
    """
    with DataStorage(database_path) as storage:
        ticker = config.ticker
        interval = config.interval.__str__()
        start_timestamp = config.start_timestamp
        end_timestamp = config.end_timestamp
        missing_timestamps = storage.get_missing_timestamps(ticker, interval, expected_timestamps)
        if len(missing_timestamps) != 0:
            min_timestamp = min(missing_timestamps)
            max_timestamp = max(missing_timestamps)
            interval_ms = int(config.interval.value.total_seconds() * 1000)
            new_config = replace(
                config,
                start_timestamp=min_timestamp,
                end_timestamp=max(max_timestamp + interval_ms, min_timestamp + interval_ms),
            )
            new_data = broker.get_bars(new_config)
            storage.add_data_to_storage(interval, new_data)

        data = storage.get_data(ticker, interval, start_timestamp, end_timestamp)

    return data