BaseProvider

Pactole Index / Data / BaseProvider

Auto-generated documentation for data.base_provider module.

BaseProvider
BaseProvider

BaseProvider

Show source in base_provider.py:31

A base class for data providers.

Environment Variables: PACTOLE_CACHE_ROOT (str): The root directory for cache files. Defaults to "pactole".

Arguments

resolver BaseResolver - An instance of a resolver to fetch archive information.
parser BaseParser - An instance of a parser to process the archive content. draw_days (DrawDays | Iterable[Day | Weekday], optional): An instance of DrawDays or an iterable of Day or Weekday representing the draw days of the lottery. Defaults to an empty tuple. draw_day_refresh_time (str | int | datetime.time, optional): The refresh threshold time on draw days. It can be provided as a string in "HH:MM" format, an integer representing the hour, or a datetime.time object. Defaults to None, which will be interpreted as 22:00 (10 PM). combination_factory (CombinationFactory | LotteryCombination | Any): A factory function or class to create a combination instance. If None, a default LotteryCombination instance will be used. Default is None.
cache_name str, optional - The name of the cache. Defaults to "default". cache_root_name (str | None, optional): The name of the root cache directory. If not provided, it will be taken from the PACTOLE_CACHE_ROOT environment variable or default to "pactole". Defaults to None.
refresh_timeout int, optional - The timeout in seconds for refreshing the cache. Defaults to 300 seconds (5 minutes).

Examples

>>> provider = BaseProvider(
...     resolver=MyResolver(),
...     parser=MyParser()
...     draw_days=[Weekday.TUESDAY, Weekday.FRIDAY],
...     draw_day_refresh_time="21:30",
...     combination_factory=EuroMillionsCombination,
...     cache_name="euromillions",
...     refresh_timeout=300,
... )
>>> provider.load()
[DrawRecord(
    period='202001',
    draw_date=date(2020, 1, 1),
    deadline_date=date(2020, 1, 15),
    combination=EuroMillionsCombination(numbers=[5, 12, 23, 34, 45], stars=[2, 9]),
    numbers={'number': [5, 12, 23, 34, 45], 'star': [2, 9]},
    winning_ranks=[WinningRank(rank=1, winners=1, gain=1000000.0), ...]
), ...]

Signature

class BaseProvider:
    def __init__(
        self,
        resolver: BaseResolver,
        parser: BaseParser,
        draw_days: DrawDays | Iterable[Day | Weekday] = (),
        draw_day_refresh_time: str | int | datetime.time | None = None,
        combination_factory: CombinationFactory | LotteryCombination | Any = None,
        cache_name: str = DEFAULT_CACHE_NAME,
        cache_root_name: str | None = None,
        refresh_timeout: int = DEFAULT_REFRESH_TIMEOUT,
    ) -> None: ...

BaseProvider()._build_cache

Show source in base_provider.py:446

Build the data file from the manifest of archives.

Signature

def _build_cache(self, manifest: Manifest) -> None: ...

BaseProvider()._check_archive_chain

Show source in base_provider.py:406

Check the chain of archives to ensure there are no gaps in the data.

Signature

def _check_archive_chain(self, manifest: Manifest) -> bool: ...

BaseProvider()._check_archives

Show source in base_provider.py:386

Check the list of archives match the list of archives from the resolver.

Signature

def _check_archives(self, manifest: Manifest) -> bool: ...

BaseProvider()._check_last_archive

Show source in base_provider.py:425

Check the last archive to ensure it is up to date with the latest draw date.

Signature

def _check_last_archive(self, manifest: Manifest) -> bool: ...

BaseProvider()._get_archive_path

Show source in base_provider.py:475

Get the file path for the archive file of a given archive name.

Signature

def _get_archive_path(self, name: str) -> Path: ...

BaseProvider()._get_source_path

Show source in base_provider.py:471

Get the file path for the source file of a given archive name.

Signature

def _get_source_path(self, name: str) -> Path: ...

BaseProvider()._load_manifest

Show source in base_provider.py:360

Load the manifest of archives.

Signature

def _load_manifest(self, force: bool) -> Manifest: ...

BaseProvider()._load_record

Show source in base_provider.py:461

Load a DrawRecord instance from a dictionary of data.

Signature

def _load_record(self, data: dict) -> DrawRecord: ...

BaseProvider()._load_record_list

Show source in base_provider.py:465

Load a list of DrawRecord instances from a list of dictionaries.

Signature

def _load_record_list(self, data: list[dict] | None) -> list[DrawRecord]: ...

BaseProvider()._load_source

Show source in base_provider.py:479

Load the archive content from the given URL and store it in the specified path.

Signature

def _load_source(self, url: str, path: Path) -> None: ...

BaseProvider()._parse_archive

Show source in base_provider.py:494

Parse the archive file and extract relevant information.

Signature

def _parse_archive(self, archive: Path) -> ArchiveContentInfo: ...

BaseProvider()._parse_source

Show source in base_provider.py:490

Parse the source file and store the results in the archive path.

Signature

def _parse_source(self, source: Path, archive: Path) -> None: ...

BaseProvider()._refresh_archive

Show source in base_provider.py:367

Refresh a specific archive.

Signature

def _refresh_archive(self, name: str, url: str, force: bool = False) -> ArchiveInfo: ...

BaseProvider()._refresh_if_needed

Show source in base_provider.py:355

Refresh the provider's cache if necessary.

Signature

def _refresh_if_needed(self, force: bool = False) -> None: ...

BaseProvider().combination_factory

Show source in base_provider.py:179

Return the combination factory used by the provider.

Returns

CombinationFactory - The combination factory used by the provider.

Examples

>>> provider = BaseProvider(
...     resolver=MyResolver(),
...     parser=MyParser(),
...     draw_days=[Weekday.MONDAY, Weekday.THURSDAY],
...     combination_factory=EuroMillionsCombination,
... )
>>> provider.combination_factory
<class 'pactole.combinations.euro_millions.EuroMillionsCombination'>
>>> provider.combination_factory()
EuroMillionsCombination(numbers=[], stars=[])

Signature

@property
def combination_factory(self) -> CombinationFactory: ...

BaseProvider().draw_day_refresh_time

Show source in base_provider.py:161

Return the refresh threshold time used on draw days.

Returns

datetime.time - The refresh threshold time used on draw days.

Examples

>>> provider = BaseProvider(
...     resolver=MyResolver(),
...     parser=MyParser(),
...     draw_day_refresh_time="21:30",
... )
>>> provider.draw_day_refresh_time
datetime.time(21, 30)

Signature

@property
def draw_day_refresh_time(self) -> datetime.time: ...

BaseProvider().draw_days

Show source in base_provider.py:137

Return the draw days of the lottery.

Returns

DrawDays - The draw days of the lottery.

Examples

>>> provider = BaseProvider(
...     resolver=MyResolver(),
...     parser=MyParser(),
...     draw_days=[Weekday.MONDAY, Weekday.THURSDAY],
... )
>>> provider.draw_days
DrawDays(days=(Weekday.MONDAY, Weekday.THURSDAY))
>>> provider.draw_days.days
(Weekday.MONDAY, Weekday.THURSDAY)
>>> provider.draw_days.get_last_draw_date(date(2024, 6, 5))
datetime.date(2024, 6, 3)
>>> provider.draw_days.get_next_draw_date(date(2024, 6, 5))
datetime.date(2024, 6, 6)

Signature

@property
def draw_days(self) -> DrawDays: ...

BaseProvider().load

Show source in base_provider.py:200

Get the cached data as a list of DrawRecord instances.

If the cache is missing or outdated, it will be refreshed before returning the data.

Arguments

force bool, optional - If True, forces a refresh of the cache before getting the data. Defaults to False.

Returns

list[DrawRecord] - A list of DrawRecord instances representing the cached data.

Examples

>>> provider = BaseProvider(
...     resolver=MyResolver(),
...     parser=MyParser(),
...     draw_days=[Weekday.TUESDAY, Weekday.FRIDAY],
...     combination_factory=EuroMillionsCombination,
...     cache_name="euromillions"
... )
>>> provider.load()
[DrawRecord(
    period='202001',
    draw_date=date(2020, 1, 1),
    deadline_date=date(2020, 1, 15),
    combination=EuroMillionsCombination(numbers=[5, 12, 23, 34, 45], stars=[2, 9]),
    numbers={'number': [5, 12, 23, 34, 45], 'star': [2, 9]},
    winning_ranks=[WinningRank(rank=1, winners=1, gain=1000000.0), ...]
), ...]

Signature

def load(self, force: bool = False) -> list[DrawRecord]: ...

BaseProvider().load_raw

Show source in base_provider.py:233

Get the cached data as a list of dictionaries.

If the cache is missing or outdated, it will be refreshed before returning the data.

The returned list of dictionaries will have the same structure as the data stored in the cache file, without being transformed into DrawRecord instances. This can be useful for debugging or for scenarios where raw data manipulation is required, like exporting to Pandas DataFrame or performing custom analyses.

Arguments

force bool, optional - If True, forces a refresh of the cache before getting the data. Defaults to False.

Returns

list[dict] - A list of dictionaries representing the cached data.

Examples

>>> provider = BaseProvider(
...     resolver=MyResolver(),
...     parser=MyParser(),
...     draw_days=[Weekday.TUESDAY, Weekday.FRIDAY],
...     combination_factory=EuroMillionsCombination,
...     cache_name="euromillions"
... )
>>> provider.load_raw()
[
    {
        'period': '202001',
        'draw_date': '2020-01-01',
        'deadline_date': '2020-01-15',
        'numbers_1': '5',
        'numbers_2': '12',
        'numbers_3': '23',
        'numbers_4': '34',
        'numbers_5': '45',
        'stars_1': '2',
        'stars_2': '9',
        ...
    },
    ...
]

Signature

def load_raw(self, force: bool = False) -> list[dict]: ...

BaseProvider().need_refresh

Show source in base_provider.py:323

Check if the cache needs to be refreshed based on the last draw date.

Rules: - When the cache is missing, it needs to be refreshed. - If the refresh timeout is active and not expired, the cache is considered still valid. - If the cache is outdated or if the cache is empty, it needs to be refreshed. - Otherwise, it checks the last draw date from the manifest of archives and compares it with the last draw date from the cache.

Returns

bool - True if the cache needs to be refreshed, False otherwise.

Signature

def need_refresh(self) -> bool: ...

BaseProvider().refresh

Show source in base_provider.py:279

Refresh the provider's cache.

If force is True, it will refresh the cache even if it is still valid. Otherwise, it will check the manifest of archives and refresh it if necessary.

The data file will be rebuilt if missing or if the manifest was refreshed.

Arguments

force bool, optional - If True, forces a refresh even if the cache is still valid. Defaults to False.

Examples

>>> provider = BaseProvider(
...     resolver=MyResolver(),
...     parser=MyParser(),
...     draw_days=[Weekday.TUESDAY, Weekday.FRIDAY],
...     combination_factory=EuroMillionsCombination,
...     cache_name="euromillions"
... )
>>> provider.refresh()

Signature

def refresh(self, force: bool = False) -> None: ...