Models

Models for NiveshPy.

OHLC

Bases: NamedTuple

Class to hold OHLC data.

get_polars_schema() classmethod

Get the Polars schema for the OHLC class.

Source code in niveshpy/models/base.py

@classmethod
def get_polars_schema(cls) -> pl.Schema:
    """Get the Polars schema for the OHLC class."""
    return pl.Schema(
        {
            "symbol": pl.String(),
            "date": pl.Date(),
            "open": pl.Decimal(scale=4),
            "high": pl.Decimal(scale=4),
            "low": pl.Decimal(scale=4),
            "close": pl.Decimal(scale=4),
        }
    )

Plugin()

Bases: ABC

Base class for all plugins.

Initialize the plugin.

Source code in niveshpy/models/plugins.py

def __init__(self) -> None:
    """Initialize the plugin."""
    super().__init__()

get_info() abstractmethod classmethod

Return plugin information.

Source code in niveshpy/models/plugins.py

@classmethod
@abc.abstractmethod
def get_info(cls) -> PluginInfo:
    """Return plugin information."""
    raise NotImplementedError("Subclasses must implement this method.")

get_sources() abstractmethod

Return a list of sources for the plugin.

Source code in niveshpy/models/plugins.py

@abc.abstractmethod
def get_sources(self) -> Iterable[Source]:
    """Return a list of sources for the plugin."""
    raise NotImplementedError("Subclasses must implement this method.")

PluginInfo(name, description, version, author, author_email)

Class to hold plugin information.

Initialize the plugin info.

Source code in niveshpy/models/plugins.py

def __init__(self, name, description, version, author, author_email):
    """Initialize the plugin info."""
    self.name = name
    self.description = description
    self.version = version
    self.author = author
    self.author_email = author_email

__repr__()

Return a string representation of the plugin info.

Source code in niveshpy/models/plugins.py

def __repr__(self):
    """Return a string representation of the plugin info."""
    return f'PluginInfo(name="{self.name}", description="{self.description}", version="{self.version}", author="{self.author}", author_email="{self.author_email}")'

Quote

Bases: NamedTuple

Class to hold price data.

get_polars_schema() classmethod

Get the Polars schema for the Quote class.

Source code in niveshpy/models/base.py

@classmethod
def get_polars_schema(cls) -> pl.Schema:
    """Get the Polars schema for the Quote class."""
    return pl.Schema(
        {
            "symbol": pl.String(),
            "date": pl.Date(),
            "price": pl.Decimal(scale=4),
        }
    )

ReturnFormat

Bases: Enum

Enum for return formats.

This enum defines the different formats in which data can be returned.

Attributes:

Name	Type	Description
DICT		Format as a dictionary mapping column names as keys.
PL_DATAFRAME		Format as a Polars DataFrame.
PL_LAZYFRAME		Format as a Polars LazyFrame. This is the internal format used by niveshpy.
PD_DATAFRAME		Format as a Pandas DataFrame.
JSON		Format as a JSON string.
CSV		Format as a CSV string.

Source()

Bases: ABC

Base class for all data sources.

Initialize the source.

Source code in niveshpy/models/sources.py

def __init__(self) -> None:
    """Initialize the source."""
    super().__init__()

get_quotes(*tickers, start_date=None, end_date=None) abstractmethod

Get the quotes for the given tickers.

Parameters:

Name	Type	Description	Default
*tickers	str	The list of tickers to get quotes for. If empty, all tickers will be fetched.	()
start_date	Optional[date]	The start date for the quotes. If none, the source should return the latest data.	None
end_date	Optional[date]	The end date for the quotes. If none, the source should return the latest data.	None

Note

• If both start_date and end_date are None, the source should return the latest data.
• If only one date is provided, the source should return data for that date.
• Date range would never exceed the data_group_period specified in the source config.

Returns:

Type	Description
QuotesIterable	An iterable of Quote or OHLC objects or a Polars DataFrame or a Pandas DataFrame.

Source code in niveshpy/models/sources.py

@abc.abstractmethod
def get_quotes(
    self,
    *tickers: str,
    start_date: date | None = None,
    end_date: date | None = None,
) -> QuotesIterable:
    """Get the quotes for the given tickers.

    Args:
        *tickers (str): The list of tickers to get quotes for. If empty, all tickers will be fetched.
        start_date (Optional[date]): The start date for the quotes. If none, the source should return the latest data.
        end_date (Optional[date]): The end date for the quotes. If none, the source should return the latest data.

        This property should be ignored if the source uses the ALL_TICKERS strategy.

    Note:
        - If both start_date and end_date are None, the source should return the latest data.
        - If only one date is provided, the source should return data for that date.
        - Date range would never exceed the `data_group_period` specified in the source config.

    Returns:
        An iterable of Quote or OHLC objects or a Polars DataFrame or a Pandas DataFrame.
    """
    raise NotImplementedError("Subclasses must implement this method.")

get_source_config() abstractmethod classmethod

Get source configuration.

Source code in niveshpy/models/sources.py

@classmethod
@abc.abstractmethod
def get_source_config(cls) -> SourceConfig:
    """Get source configuration."""
    raise NotImplementedError("Subclasses must implement this method.")

get_source_info() abstractmethod classmethod

Get source information.

Source code in niveshpy/models/sources.py

@classmethod
@abc.abstractmethod
def get_source_info(cls) -> SourceInfo:
    """Get source information."""
    raise NotImplementedError("Subclasses must implement this method.")

get_source_key() abstractmethod classmethod

Get a unique key for this source.

Source code in niveshpy/models/sources.py

@classmethod
@abc.abstractmethod
def get_source_key(cls) -> str:
    """Get a unique key for this source."""
    raise NotImplementedError("Subclasses must implement this method.")

get_tickers() abstractmethod

Get the list of tickers.

Source code in niveshpy/models/sources.py

@abc.abstractmethod
def get_tickers(self) -> TickersIterable:
    """Get the list of tickers."""
    raise NotImplementedError("Subclasses must implement this method.")

SourceConfig

Bases: NamedTuple

Class to hold source configuration.

data_group_period = None class-attribute instance-attribute

The time period for which data can be grouped at source.

This is used to limit the amount of calls made to the source. For example, if the source can return data for 1 month at a time, this should be set to 30 days.

If this value is None, the data will not be grouped.

This value will be passed to polars.DataFrame.group_by_dynamic to group the data by the specified time period.

Default is None.

data_refresh_interval = timedelta(days=1) class-attribute instance-attribute

The time interval at which the source can be checked for new data.

Note that this only applies to new data. Historical data, once fetched, will not be fetched again.

This frequency will be ticker-specific unless the source uses the ALL_TICKERS strategy, in which case it will be source-specific.

source_strategy = SourceStrategy.DEFAULT class-attribute instance-attribute

The strategy to use for the source. Multiple strategies can be combined using bitwise OR. This is used to determine how to fetch and store data from the source.

ticker_refresh_interval = None class-attribute instance-attribute

The time interval at which the source can be checked for new tickers.

If this value is None, the source will not be checked for new tickers. Default is None.

SourceInfo

Bases: NamedTuple

Class to hold source information.

version instance-attribute

This will later add support for source migrations.

SourceStrategy

Bases: Flag

Enum for source strategies.

This enum defines the different strategies for data sources. Strategies can be combined using bitwise OR. These help in determining how to fetch and store data from the source. For example, if the source only supports fetching data for all tickers at a time, NiveshPy will store the data and use them in the future automatically.

ALL_TICKERS = auto() class-attribute instance-attribute

The source fetches data for all tickers at once. Used for sources that do not support fetching data only for a list of provided tickers.

DEFAULT = 0 class-attribute instance-attribute

Use this when the source does not require any special strategy.

Default strategy: - The source only fetches data for the provided tickers (or all tickers if none are provided). - The source returns OHLC data.

SINGLE_QUOTE = auto() class-attribute instance-attribute

The source returns a single quote for the provided ticker. Used for sources that do not support fetching OHLC data.

Ticker

Bases: NamedTuple

Class to hold ticker information.

get_polars_schema() classmethod

Get the Polars schema for the Ticker class.

Source code in niveshpy/models/base.py

@classmethod
def get_polars_schema(cls) -> pl.Schema:
    """Get the Polars schema for the Ticker class."""
    return pl.Schema(
        {
            "symbol": pl.String(),
            "name": pl.String(),
            "isin": pl.String(),
        }
    )