finagg package

Subpackages

Submodules

finagg.backend module

finagg configuration and global SQLAlchemy setup. Backend file paths and SQLAlchemy engine database URLs are configured in this module at runtime according to environment variables.

Environment variables should ideally be configured using an .env file in the desired working directory. Running finagg install will automaticaly setup the .env file for you according to your input values. Environment variables assigned in the .env file are loaded on the finagg module’s first instantiation.

finagg.backend.root_path

Parent directory of the findata directory where the backend database and API cache file will be stored (unless otherwise configured according to the relevant environment variables). This can be set with the FINAGG_ROOT_PATH environment variable. This defaults to and is typically set to the current working directory. It’s recommended you permanently set this value using the finagg install CLI.

finagg.backend.disable_http_cache

Whether the disable the HTTP requests cache. Instead of a cachable session, a default, uncached user session will be used for all requests.

finagg.backend.http_cache_path

Path to the API cache file. This can be set with the FINAGG_HTTP_CACHE_PATH environment variable and should NOT include a file extension. All API implementations share the same cache backend.

finagg.backend.database_path

Default path to the database file. The FINAGG_DATABASE_URL environment variable will take precedence over this value.

finagg.backend.database_url

SQLAlchemy URL to the database. This can be set with the FINAGG_DATABASE_URL environment variable and should include a file extension. This defaults to f"sqlite:///{finagg.backend.database_path}".

finagg.backend.engine

The default SQLAlchemy engine for the backend database. All feature and SQL submodules use this engine and the database URL as configured by database_url for reading and writing to and from the database by default.

finagg.frame module

datetime but for fiscal frames (quarter and year pairs).

finagg.frame.is_valid_fiscal_seq(seq: list[int], /) → bool[source]

Determine if the sequence of fiscal quarter differences is continuous.

A sequence that contains a jump from Q3 to Q1 is considered valid because companies often aren’t required (and don’t) report fundamentals/metrics for Q4.

Parameters:: seq – Sequence of fiscal quarter differences.
Returns:: Whether the sequence is a valid, ordered fiscal quarter sequence.

Examples

This fiscal quarter difference sequence is synonymous with the quarter sequence [Q2, Q3, Q1, Q2, Q3].

>>> from finagg.frame import is_valid_fiscal_seq
>>> is_valid_fiscal_seq([1, 2, 1, 1])
True

This fiscal quarter difference sequence is synonymous with the quarter sequence [Q2, Q3, Q4, Q1, Q2].

>>> is_valid_fiscal_seq([1, 1, 1, 1])
True

This fiscal quarter difference sequence is synonymous with the quarter sequence [Q1, Q4] which is not a valid fiscal sequence.

>>> is_valid_fiscal_seq([3])
False

class finagg.frame.FiscalDelta(years: int = 0, quarters: int = 0)[source]

Bases: object

A displacement or change from a FiscalFrame.

Examples

Get the total number of quarters from a fiscal delta.

>>> from finagg.frame import FiscalDelta
>>> delta = FiscalDelta(2, 2)
>>> int(delta)
10

years: int = 0: Year delta.

quarters: int = 0: Quarter delta.

class finagg.frame.FiscalFrame(year: int, quarter: int)[source]

Bases: object

A year and quarter pair.

Useful for comparing fiscal frames or getting fiscal frames based on fiscal deltas (changes in fiscal years or quarters).

Examples

Getting the fiscal frame a couple years and quarters ahead.

>>> from finagg.frame import FiscalDelta, FiscalFrame
>>> frame = FiscalFrame(1995, 1)
>>> frame + FiscalDelta(2, 2)
FiscalFrame(year=1997, quarter=3)

Adding/subtracting with integers assumes integers are quarters.

>>> frame + 2
FiscalFrame(year=1995, quarter=3)

Adding/subtracting with tuples converts tuples to FiscalDelta.

>>> frame + (2, 2)
FiscalFrame(year=1997, quarter=3)

Getting quarter differences between frames and determining if the sequence is a valid, ordered quarterly sequence.

>>> from finagg.frame import FiscalFrame, is_valid_fiscal_seq
>>> df = finagg.sec.api.company_concept.get("AssetsCurrent", ticker="AAPL")
>>> df = finagg.sec.api.filter_original_filings(df, form="10-Q", units="USD")
>>> frames: pd.Series = df["fy"].astype(int).astype(str) + df["fp"].astype(str)
>>> frames = frames.apply(lambda row: FiscalFrame.fromstr(row))
>>> frames = frames.diff(periods=1).dropna().astype(int)
>>> is_valid_fiscal_seq(frames.tolist())
True

year: int: Fiscal year (e.g., 1995).

quarter: int: Fiscal quarter (i.e., 1, 2, 3, 4).

classmethod fromstr(s: str, /) → FiscalFrame[source]: Split a string into year-quarter parts by splitting on alphabetical characters.

finagg.portfolio module

Definitions related to tracking an investment portfolio of cash and stocks. Underlying arithmetic uses exact decimal representations for max precision.

class finagg.portfolio.Position(cost: float, quantity: float, /)[source]

Bases: object

A position in holding a security.

Parameters:

cost – Initial purchase cost.
quantity – Number of shares held at cost.

property average_cost_basis: float: Average dollar cost for each share in the position.

buy(cost: float, quantity: float, /) → float[source]

Buy quantity of the position for cost.

Parameters:

cost – Cost to buy at.
quantity – Number of shares to buy.

Returns:

Value of the bought position.

Examples

>>> from finagg.portfolio import Position
>>> pos = Position(100.0, 1)
>>> pos.buy(50.0, 1)
50.0
>>> pos.total_cost_basis
150.0
>>> pos.average_cost_basis
75.0
>>> pos.quantity
2.0

property quantity: float: Current number of shares owned in the position.

sell(cost: float, quantity: float, /) → float[source]

Sell quantity of the position for cost.

Parameters:

cost – Cost to sell at.
quantity – Number of shares to sell.

Returns:

Value of the sold position.

Raises:

ValueError – If there aren’t enough shares to sell in the position.

Examples

>>> from finagg.portfolio import Position
>>> pos = Position(100.0, 2)
>>> pos.sell(50.0, 1)
50.0
>>> pos.total_cost_basis
100.0
>>> pos.average_cost_basis
100.0
>>> pos.quantity
1.0

property total_cost_basis: float: Total dollar cost for all shares in the position. The amount of dollars or cash invested in this security.

total_dollar_change(cost: float, /) → float[source]

Compute the total dollar change relative to the average cost basis and the current value of the security.

Parameters:: cost – Current value of one share.
Returns:: Total dollar change in value.

Examples

>>> from finagg.portfolio import Position
>>> pos = Position(100.0, 1)
>>> pos.total_dollar_change(50.0)
-50.0

total_log_change(cost: float, /) → float[source]

Compute the total log change relative to the average cost basis and the current value of the security.

Parameters:: cost – Current value of one share.
Returns:: Total log change in value. Negative indicates loss in value, positive indicates gain in value.

Examples

>>> from finagg.portfolio import Position
>>> pos = Position(100.0, 1)
>>> pos.total_log_change(50.0)
-0.6931471805599453

total_percent_change(cost: float, /) → float[source]

Compute the total percent change relative to the average cost basis and the current value of the security.

Parameters:: cost – Current value of one share.
Returns:: Total percent change in value. Negative indicates loss in value, positive indicates gain in value.

Examples

>>> from finagg.portfolio import Position
>>> pos = Position(100.0, 1)
>>> pos.total_percent_change(50.0)
-0.5

class finagg.portfolio.Portfolio(cash: float, /)[source]

Bases: object

A collection of cash and security positions.

Parameters:: cash – Starting cash position.

positions: dict[str, finagg.portfolio.Position]: Existing positions for each security.

property cash: float: Total liquid cash on-hand.

buy(symbol: str, cost: float, quantity: float, /) → float[source]

Buy quantity of security with symbol for cost.

Parameters:

symbol – Security ticker.
cost – Cost to buy the symbol at.
quantity – Number of shares to purchase.

Returns:

Value of the symbol’s bought position in the portfolio.

Raises:

ValueError – If the portfolio doesn’t have enough cash to execute the buy order.

Examples

>>> from finagg.portfolio import Portfolio
>>> port = Portfolio(1000.0)
>>> port.buy("AAPL", 100.0, 1)
100.0
>>> pos = port["AAPL"]
>>> pos.total_cost_basis
100.0
>>> pos.average_cost_basis
100.0
>>> pos.quantity
1.0

deposit(cash: float, /) → float[source]

Deposit more cash into the portfolio.

Parameters:: cash – Cash to deposit.
Returns:: Total cash in the portfolio.

sell(symbol: str, cost: float, quantity: float, /) → float[source]

Sell quantity of security with symbol for cost.

Parameters:

symbol – Security ticker.
cost – Cost to sell the symbol at.
quantity – Number of shares to sell.

Returns:

Value of the symbol’s sold position in the portfolio.

Examples

>>> from finagg.portfolio import Portfolio
>>> port = Portfolio(1000.0)
>>> port.buy("AAPL", 100.0, 2)
200.0
>>> port.sell("AAPL", 50.0, 1)
50.0
>>> pos = port["AAPL"]
>>> pos.total_cost_basis
100.0
>>> pos.average_cost_basis
100.0
>>> pos.quantity
1.0

property total_deposits: float: Total cash deposited since starting the portfolio.

total_dollar_change(costs: dict[str, float], /) → float[source]

Compute the total dollar change relative to the total deposits made into the portfolio.

Parameters:: costs – Mapping of symbol to its current value of one share.
Returns:: Total dollar change in value.

Examples

>>> from finagg.portfolio import Portfolio
>>> port = Portfolio(1000.0)
>>> port.buy("AAPL", 100.0, 1)
100.0
>>> port.total_dollar_change({"AAPL": 50.0})
-50.0

total_dollar_value(costs: dict[str, float], /) → float[source]

Compute the total dollar value of the portfolio.

Parameters:: costs – Mapping of symbol to its current value of one share.
Returns:: Total dollar value.

Examples

>>> from finagg.portfolio import Portfolio
>>> port = Portfolio(1000.0)
>>> port.buy("AAPL", 100.0, 1)
100.0
>>> port.total_dollar_value({"AAPL": 50.0})
950.0

total_log_change(costs: dict[str, float], /) → float[source]

Compute the total log change relative to the total deposits made into the portfolio.

Parameters:: costs – Mapping of symbol to its current value of one share.
Returns:: Total log change in value. Negative indicates loss in value, positive indicates gain in value.

Examples

>>> from finagg.portfolio import Portfolio
>>> port = Portfolio(1000.0)
>>> port.buy("AAPL", 100.0, 1)
100.0
>>> port.total_log_change({"AAPL": 50.0})
-0.051293294387550536

total_percent_change(costs: dict[str, float], /) → float[source]

Compute the total percent change relative to the total deposits made into the portfolio.

Parameters:: costs – Mapping of symbol to its current value of one share.
Returns:: Total percent change in value. Negative indicates loss in value, positive indicates gain in value.

Examples

>>> from finagg.portfolio import Portfolio
>>> port = Portfolio(1000.0)
>>> port.buy("AAPL", 100.0, 1)
100.0
>>> port.total_percent_change({"AAPL": 50.0})
-0.05

property total_withdrawals: float: Total cash withdrawn since starting the portfolio.

withdraw(cash: float, /) → float[source]

Withdraw cash from the portfolio.

Parameters:: cash – Cash to withdraw.
Returns:: Total cash in the portfolio.
Raises:: ValueError – If the portfolio doesn’t have at least cash liquid cash to withdraw.

Examples

>>> from finagg.portfolio import Portfolio
>>> port = Portfolio(1000.0)
>>> port.withdraw(100.0)
900.0

finagg.ratelimit module

Customizable rate-limiting for requests-style getters.

The definitions within this submodule are used throughout finagg for respecting 3rd party API rate limits to avoid server-side throttling.

class finagg.ratelimit.RateLimit(limit: float, period: float | timedelta, /, *, buffer: float = 0.0)[source]

Bases: ABC

Interface for defining a rate limit for an external API getter.

You can create a custom rate-limiter by inheriting from this class and implementing a custom eval() method.

Parameters:

limit – Max limit within period (e.g., max number of requests, errors, size in memory, etc.).
period – Time interval for evaluating limit.
buffer – Reduce limit by this fraction. Adds a bit of leeway to ensure limit is not reached. Useful for enforcing response size limits.

finagg.testing module

Testing utils used for finagg’s own unit tests.

finagg.testing.sqlite_engine(path: str, /, *, metadata: None | MetaData = None, table: None | Table = None) → Generator[Engine, None, None][source]

Yield a test database engine that’s cleaned-up after usage.

Parameters:

path – Path to SQLite database file.
metadata – Optional metadata for creating and dropping tables before and after yielding the engine, respectively.
table – Optional table for creating and dropping before and after yielding the engine, respectively.

Returns:

A database engine that’s subsequently disposed of and whose respective database file is deleted after use.

Raises:

ValueError – If both metadata and table are provided.

Examples

Using the testing util as a pytest fixture.

>>> import pytest
>>> from sqlalchemy.engine import Engine
>>> @pytest.fixture
... def engine() -> Engine:
...     yield from finagg.testing.sqlite_engine("/path/to/db.sqlite")

finagg.utils module

Generic utils used by subpackages.

finagg.utils.CamelCase(s: str, /) → str[source]

Transform a string to CamelCase.

Credit:: https://stackoverflow.com/a/1176023

Parameters:: s – Any string.
Returns:: A string in CamelCase format.

Examples

>>> finagg.utils.CamelCase("snakes_are_dope") == "SnakesAreDope"
True
>>> finagg.utils.CamelCase("bar") == "Bar"
True

finagg.utils.expand_csv(values: str | list[str], /) → set[str][source]

Expand the given list of strings into a set of strings, where each value in the list of strings could be:

Comma-separated values

A path that points to a CSV file containing values

A regular ol’ string

Parameters:: values – List of strings denoting comma-separated values, or CSV files containing comma-separated values.
Returns:: A set of all strings found within the given list.

Examples

>>> ts = finagg.utils.expand_csv(["AAPL,MSFT"])
>>> "AAPL" in ts
True

finagg.utils.get_func_cols(table: Table | DataFrame, /) → list[str][source]

Return the column names in table that have the format FUNC(arg0, arg1, ...).

Parameters:: table – SQLAlchemy table or dataframe.
Returns:: List of functional-style column names in table. Returns an empty list if none are found.
Raises:: TypeError – If the given object is not a SQLAlchemy table or dataframe.

finagg.utils.parse_func_call(s: str, /) → None | tuple[str, list[str]][source]

Parse a function’s name and its arguments’ names from a string of format FUNC(arg0, arg1, ...).

Parameters:: s – Any string of format FUNC(arg0, arg1, ...).
Returns:: A tuple containing the parsed function’s name and its arguments’ names. Returns None if the string doesn’t match the expected format.

Examples

>>> finagg.utils.parse_func_call("LOG_CHANGE(high, open)")
('LOG_CHANGE', ['high', 'open'])

finagg.utils.resolve_col_order(table: Table, df: DataFrame, /, *, extra_ignore: None | list[str] = None) → DataFrame[source]

Reorder the columns in df to match the order of the columns in table.

Parameters:

table – SQLAlchemy table that defines the column order. Primary keys are ignored from the column order as they’re assumed to be used as part of the index in df.
df – Dataframe to reorder.
extra_ignore – Extra columns to ignore in the reordering. Sometimes columns aren’t used as primary keys but are used as part of the index in the dataframe. Those columns should be provided in this option.

Returns:

Dataframe with columns ordered according to the column order in table.

finagg.utils.resolve_func_cols(table: Table, df: DataFrame, /, *, drop: bool = False, inplace: bool = False) → DataFrame[source]

Inspect table and apply functions to columns that exist in table and df according to columns named like FUNC(col0, col1, ...) within table such that new columns in df are the result of the applied functions and have names matching the function call signatures.

Parameters:

table – SQLAchemy table that defines a superset of columns that should exist in df.
df – Dataframe that contains a subset of columns within table that will be updated with columns defined by table that have names like FUNC(col0, col1, ...).
drop – Whether to drop all other columns on the returned dataframe except for the columns in table.
inplace – Whether to perform operations in-place and use df as the output dataframe.

Returns:

A new dataframe with columns from df and columns according to columns named within table like FUNC(col0, col1, ...) where columns col0 and col1 exist in df.

Raises:

ValueError – If the function parsed from the column name has no supported and corresponding function.

finagg.utils.safe_log_change(series: Series, other: None | Series = None) → Series[source]

Safely compute log change between two columns.

Replaces Inf values with NaN and forward-fills. This function is meant to be used with pd.Series.apply.

Parameters:

series – Series of values.
other – Reference series to compute change against. Defaults to series shifted forward one index.

Returns:

A series representing percent changes of col.

finagg.utils.safe_pct_change(series: Series, other: None | Series = None) → Series[source]

Safely compute percent change between two columns.

Replaces Inf values with NaN and forward-fills. This function is meant to be used with pd.Series.apply.

Parameters:

series – Series of values.
other – Reference series to compute change against. Defaults to series shifted forward one index.

Returns:

A series representing percent changes of col.

finagg.utils.setenv(name: str, value: str, /, *, exist_ok: bool = False) → Path[source]

Set the value of the environment variable name to value.

The environment variable is permanently set in the environment and in the current process.

Parameters:

name – Environment variable name.
value – Environment variable value.
exist_ok – Whether it’s okay if an environment variable of the same name already exists. If True, it will be overwritten.

Returns:

Path to the file the environment variable was written to.

Raises:

RuntimeError – If exist_ok is False and an environment variable of the same name already exists.

finagg.utils.snake_case(s: str, /) → str[source]

Transform a string to snake_case.

Credit:: https://stackoverflow.com/a/1176023

Parameters:: s – Any string.
Returns:: A string in snake_case format.

Examples

>>> finagg.utils.snake_case("CamelsAreCool") == "camels_are_cool"
True
>>> finagg.utils.snake_case("Foo") == "foo"
True

finagg.utils.today

Today’s date. Used by a number of submodules as the default end date when getting data from APIs or SQL tables.

Module contents

Main package interface.