src.irs_asset_fifo_calculator.calculate_taxes

Calculate IRS capital gains taxes using FIFO method.

This module implements IRS-style FIFO cost-basis calculations for stocks and other assets. Each purchase creates a “lot” stored in a per-asset FIFO queue. Each sale consumes lots from oldest to newest, allocating cost basis and proceeds proportionally and emitting Form 8949 rows.

This program uses a CSV file as input. This file is called “asset_tx.csv” in the published example, but any name can be used, using this name in the python call. Expected input CSV columns (at minimum):

Tx Index, Date, Asset, Amount (asset), Sell price ($), Buy price ($), Type

Additional columns such as Account number, Entity, Notes or Remaining may be present but are ignored by this module.

Key steps:

  1. Group rows by Tx Index into logical transaction blocks.

  2. For each block, classify buy/sell/fee rows and compute: buy data, sell data, and fee data.

  3. Update per-asset FIFO queues and append realized sales to a Form 8949 list.

  4. Write the Form 8949 list to “form8949.csv”.

For a full worked example, see the FIFO overview section of the docs.

Functions

is_fee(asset)

Check if asset is a fee transaction.

is_finite_number(x)

Return True if x is a finite (non-NaN, non-infinite, non-bool) real number.

main([argv])

Run the FIFO capital-gains pipeline on a CSV file.

parse_amount(value)

Parse amount from input.

parse_args([argv])

Parse args from command line.

parse_buy_and_sell(is_buy, block_type, rows, ...)

Extract the buy or sell side from a block of related transactions.

parse_row_data(block_type, rows)

Extract the necessary values from row data.

record_sale(form8949, asset, amount, ...)

Record a sale.

reduce_fifo(form8949, sell_amount, asset, ...)

Update FIFO lots for a sale.

run_fifo_pipeline(df)

Run the FIFO capital-gains pipeline on a transactions DataFrame.

update_fifo(buy_data, sell_data, fee_data, ...)

Updates FIFO dict of deque using info from this block of transactions.

Classes

AssetData(asset, amount, price, total, tx_date)

Single transaction data row for a particular asset.

FifoLot

Single FIFO lot for an asset.
class src.irs_asset_fifo_calculator.calculate_taxes.AssetData(asset: str | None, amount: float, price: float, total: float, tx_date: date)

Bases: object

Single transaction data row for a particular asset.

asset

The asset from the transaction.

Type:

str

amount

The amount of the asset in this transaction.

Type:

float

price

Unit price in USD.

Type:

float

total

Total amount. This can be cost or proceeds and is typically the amount * price +- fee_amount * fee_price

Type:

float

tx_date

Date of the transaction.

Type:

date

class src.irs_asset_fifo_calculator.calculate_taxes.FifoLot

Bases: TypedDict

Single FIFO lot for an asset.

amount

Remaining asset quantity.

Type:

float

price

Unit price in USD.

Type:

float

cost

Total cost basis in USD.

Type:

float

tx_date

Acquisition date of this lot.

Type:

date

src.irs_asset_fifo_calculator.calculate_taxes.is_fee(asset: str | None) → bool

Check if asset is a fee transaction.

In order to be a fee, the asset must start with the letters “fee”, and be longer than 3 characters.

src.irs_asset_fifo_calculator.calculate_taxes.is_finite_number(x: object) → bool

Return True if x is a finite (non-NaN, non-infinite, non-bool) real number.

src.irs_asset_fifo_calculator.calculate_taxes.main(argv: Sequence[str] | None = None) → None

Run the FIFO capital-gains pipeline on a CSV file.

This function produces an IRS Form 8949-style output file.

It reads input_file_path (by default “asset_tx.csv”), keeping only the necessary columns:

"Tx Index", "Date", "Asset", "Amount (asset)", "Sell price ($)", "Buy price ($)", "Type"

It then parses the rows, updates the FIFO ledger, and writes all sales to output_file_path (by default “form8949.csv”).

Parameters:

  • input_file_path – Path to the input CSV with raw transactions.

  • output_file_path – Path where the Form 8949-style CSV will be written.

Returns:

None.

src.irs_asset_fifo_calculator.calculate_taxes.parse_amount(value: Any) → float

Parse amount from input. Can be string or numeric. Extra whitespace is valid, $ or € signs are not.

src.irs_asset_fifo_calculator.calculate_taxes.parse_args(argv: Sequence[str] | None = None) → tuple[str, str]

Parse args from command line.

src.irs_asset_fifo_calculator.calculate_taxes.parse_buy_and_sell(is_buy: bool, block_type: Literal['Buy', 'Sell', 'Exchange', 'Transfer'], rows: DataFrame, fee_assets: set[str], fee_rows: list[int]) → tuple[str | None, float, float, float]

Extract the buy or sell side from a block of related transactions.

For non-transfer blocks, this scans the rows (excluding fee rows) to find the one representing the buy or sell side based on the sign of Amount (asset) (is_buy = True: amount > 0, is_buy = False: amount < 0). It returns that row’s asset symbol, signed amount, and unit price. If the resulting asset is in fee_assets, any fee rows are added to the returned amount. The price for USD is always forced to 1.0.

Parameters:

  • is_buy (bool) – are we parsing buy side?

  • block_type (BlockType) – Type of block ("Exchange", "Buy", "Sell", or "Transfer")

  • rows (pd.DataFrame) – the transactions for this block. Must include at least "Asset", "Amount (asset)", "Buy price ($)", and "Sell price ($)"

  • fee_assets (set[str]) – assets that have associated fee rows (e.g. "USD", not "feeUSD")

  • fee_rows (list[int]) – indices within rows that correspond to fee transactions

Returns:

A 4-tuple containing:

  • asset: the buy or sell asset

  • amount: the buy or sell amount

  • price: the buy or sell price

  • cost_or_proceeds: the cost if buy or proceeds if sell

where

  • amount keeps the sign of the transaction (positive for buys, negative for sells).

  • price is the unit price in USD (forced to 1.0 for USD).

  • cost_or_proceeds is:

    • total cost (positive) for buys, including all relevant fees.

    • total proceeds (positive or negative with fee adjustments) for sells.

Return type:

tuple[str, float, float, float]

Raises:

  • ValueError – If more than one non-fee row matches the requested side.

  • ValueError – If the fee asset is the same as the buy or sell asset but the prices are different.

Notes

  • The price for USD is always forced to 1.0.

  • Returns (None, 0.0, 0.0, 0.0) when:

    • block_type == "Transfer", or

    • no matching non-fee row is found.

Example

>>> import pandas as pd
>>> from datetime import date
>>> from calculate_taxes import parse_buy_and_sell
>>> rows = pd.DataFrame([
...     {"Tx Date": date(2024, 9, 4), "Asset": "TSLA",
...      "Amount (asset)": -25.0, "Sell price ($)": 50.0,
...      "Buy price ($)": float("nan"), "Type": "Exchange"},
...     {"Tx Date": date(2024, 9, 4), "Asset": "NVDA",
...      "Amount (asset)": 10.0, "Sell price ($)": float("nan"),
...      "Buy price ($)": 125.0, "Type": "Exchange"},
...     {"Tx Date": date(2024, 9, 4), "Asset": "feeUSD",
...      "Amount (asset)": -10.0, "Sell price ($)": 1.0,
...      "Buy price ($)": float("nan"), "Type": "Exchange"},
... ])
>>> fee_rows = [2]
>>> fee_assets = set("USD")
>>> parse_buy_and_sell(True, "Exchange", rows, fee_assets, fee_rows)
('NVDA', 10.0, 125.0, 1260.0)
>>> parse_buy_and_sell(False, "Exchange", rows, fee_assets, fee_rows)
('TSLA', -25.0, 50.0, 1240.0)
src.irs_asset_fifo_calculator.calculate_taxes.parse_row_data(block_type: Literal['Buy', 'Sell', 'Exchange', 'Transfer'], rows: DataFrame) → tuple[AssetData, AssetData, AssetData]

Extract the necessary values from row data.

Parameters:

  • block_type (BlockType) – The type of block to extract from. Can take the following values: ["Buy", "Sell", "Exchange", "Transfer"]

  • rows (pd.DataFrame) – The row data to extract from. The mandatory columns are: ["Tx Index", "Tx Date", "Asset", "Amount (asset)", "Buy price ($)", "Sell price ($)", "Type"]

Returns:

A 3-tuple with the following:

  • buy data

  • sell data

  • fee data

Return type:

tuple[AssetData, AssetData, AssetData]

Notes

  • Transfer fees are not deducted although if paid with an asset, the conversion of the asset to USD is taxed.

  • For transfers, there is no buy data. The fee data becomes the sell data.

  • Proceeds from fee assets are:

    • Added to cost if the fee asset is the same as the bought asset

    • Deducted from proceeds if the fee asset is the same as the sold asset

    • Recorded as a sale if it is a transfer or if they are different from both assets in a buy/sell/exchange.

  • Here we assume that there can only be a maximum of 1 fee asset besides the buy and sell assets. Sell and fee amount are negative in general.

  • If the fee asset is the same as the buy or sell asset, it is included in these, and the fee amount for that asset is set to 0. If there are no other fee assets, then the fee asset will be None.

  • With large enough fees, the buy amount may become negative, in which case it will later be used to update FIFO (reduce and append to form8949) rather than append to FIFO.

Example

>>> import pandas as pd
>>> from datetime import date
>>> from calculate_taxes import parse_row_data
>>> block_type = 'Buy'
>>> rows = pd.DataFrame({
...     'Tx Index': [0] * 3, 'Tx Date': [date(2024, 9, 4)] * 3,
...     'Asset': ['USD', 'NVDA', 'feeUSD'],
...     'Amount (asset)': [-1250, 10, -10],
...     'Sell price ($)': [1, 'NaN', 1],
...     'Buy price ($)': [1, 125, 1],
...     'Type': ['Buy'] * 3})
>>> buy_data, sell_data, fee_data = parse_row_data(block_type, rows)
>>> buy_data
AssetData(asset='NVDA', amount=10.0, price=125.0, total=1260.0,
tx_date=datetime.date(2024, 9, 4))
>>> sell_data
AssetData(asset='USD', amount=-1260.0, price=1.0, total=1240.0,
tx_date=datetime.date(2024, 9, 4))
>>> fee_data
AssetData(asset=None, amount=0.0, price=0.0, total=-0.0,
tx_date=datetime.date(2024, 9, 4))
src.irs_asset_fifo_calculator.calculate_taxes.record_sale(form8949: list[dict[str, str]], asset: str, amount: float, proceeds: float, cost_basis: float, acquisition_date: date, sale_date: date) → None

Record a sale.

This takes various data about the sale and appends the data to the open Form 8949 file object.

Parameters:

  • form8949 (list[dict[str, str]]) – Form 8949 list of dicts holding txs.

  • asset (str) – The asset name.

  • amount (float) – The amount of the asset units.

  • proceeds (float) – The gross dollar proceeds from this portion of the sale.

  • cost_basis (float) – The dollar cost basis for this portion of the asset, including purchase fees.

  • acquisition_date (date) – The acquisition date.

  • sale_date (date) – The sale date.

Returns:

None.

Example

>>> from calculate_taxes import record_sale
>>> from datetime import date
>>> form8949 = list()
>>> form8949.append({"Description": "10.00000000 NVDA",
...     "Date Acquired": "11/28/1982",
...     "Date Sold": "12/31/2024",
...     "Proceeds": "10000",
...     "Cost Basis": "1000",
...     "Gain or Loss": "9000"})
>>> record_sale(form8949, "TSLA", 10, 100, 90, date(2024,1,1),
...     date(2024,12,31))
>>> len(form8949)
2
>>> form8949[1]["Description"]
'10.00000000 TSLA'
>>> form8949[1]["Date Acquired"]
'01/01/2024'
>>> form8949[1]["Date Sold"]
'12/31/2024'
>>> form8949[1]["Proceeds"]
'100.00'
>>> form8949[1]["Cost Basis"]
'90.00'
>>> form8949[1]["Gain or Loss"]
'10.00'
src.irs_asset_fifo_calculator.calculate_taxes.reduce_fifo(form8949: list[dict[str, str]], sell_amount: float, asset: str, fifo_asset: deque[FifoLot], proceeds: float, sale_date: date) → None

Update FIFO lots for a sale.

This is where the FIFO cost-basis math happens. Given a sale of sell_amount units of asset, the function walks the existing lots in fifo_asset from oldest to newest, consuming quantities until the sale is fully matched. For each lot (or partial lot) that is used, it:

  1. Computes the fraction of the lot that is being sold.

  2. Allocates the same fraction of the lot’s cost to this sale.

  3. Allocates a proportional share of the sale’s total proceeds based on how many units from the lot are used.

  4. Records a Form 8949 row via record_sale().

  5. Updates or removes the lot from fifo_asset to reflect what remains after the sale.

In other words, the earliest purchases (oldest lots) are always matched first, which is exactly the FIFO (First In, First Out) method required for these cost-basis calculations.

Example

Suppose you bought 10 NVDA at $10 (cost $100) and later 5 NVDA at $11 (cost $55), then sell 12 NVDA for total proceeds of $144. The FIFO matching is:

  • First 10 units come from the $10 lot

  • Remaining 2 units come from the $11 lot

This function will: * Create one Form 8949 row for the 10-unit slice of the first lot * Create another row for the 2-unit slice of the second lot * Leave 3 units in the second lot in fifo_asset

Parameters:

  • form8949 (list[dict[str, str]]) – Form 8949 list of dicts holding txs.

  • sell_amount (float) – this sale’s amount

  • asset (str) – this asset

  • fifo_asset (deque[FifoLot]) – purchases for this token defined by their amount, price, cost, and date

  • proceeds (float) – this sale’s proceeds

  • sale_date (date) – this sale’s date

Returns:

None

Example

>>> from calculate_taxes import reduce_fifo
>>> from datetime import date
>>> from collections import defaultdict, deque
>>> form8949 = list()
>>> fifo = defaultdict(deque)
>>> fifo['NVDA'].append({"amount": 10, "price": 10,
...     "cost": 100*1.002, "tx_date": date(2024, 1, 1)})
>>> fifo['NVDA'].append({"amount": 20, "price": 11,
...     "cost": 210*1.002, "tx_date": date(2024, 2, 1)})
>>> reduce_fifo(form8949, 15, 'NVDA', fifo['NVDA'], 135,
...     date(2024, 3, 1))
>>> len(fifo['NVDA'])
1
>>> abs(fifo['NVDA'][0]['amount'] - 15) < 0.001
True
>>> abs(fifo['NVDA'][0]['price'] - 11) < 0.001
True
>>> abs(fifo['NVDA'][0]['cost'] - 157.5*1.002) < 0.001
True
>>> fifo['NVDA'][0]['tx_date']
datetime.date(2024, 2, 1)
src.irs_asset_fifo_calculator.calculate_taxes.run_fifo_pipeline(df: DataFrame) → list[dict[str, str]]

Run the FIFO capital-gains pipeline on a transactions DataFrame.

The input DataFrame must contain at least the following columns:

  • "Date"

  • "Tx Index"

  • "Asset"

  • "Amount (asset)"

  • "Sell price ($)"

  • "Buy price ($)"

  • "Type"

This function is pure with respect to IO: it does not read or write any files. It returns a list of dictionaries representing rows for an IRS Form 8949-style output.

Parameters:

df (pd.DataFrame) – Raw transaction DataFrame with the columns described above.

Returns:

A list of Form 8949 rows (dicts with keys: "Description", "Date Acquired", "Date Sold", "Proceeds", "Cost Basis", "Gain or Loss").

Return type:

list[dict[str, str]]

Example

>>> import pandas as pd
>>> from calculate_taxes import run_fifo_pipeline
>>> df = pd.DataFrame([
...     # buy block (Tx Index 0)
...     {"Date": "2024-09-04", "Tx Index": 0, "Asset": "USD",
...      "Amount (asset)": -1250.0, "Sell price ($)": 1.0,
...      "Buy price ($)": 1.0, "Type": "Buy"},
...     {"Date": "2024-09-04", "Tx Index": 0, "Asset": "NVDA",
...      "Amount (asset)": 10.0, "Sell price ($)": float('nan'),
...      "Buy price ($)": 125.0, "Type": "Buy"},
...     {"Date": "2024-09-04", "Tx Index": 0, "Asset": "feeUSD",
...      "Amount (asset)": -10.0, "Sell price ($)": 1.0,
...      "Buy price ($)": float('nan'), "Type": "Buy"},
...     # sell block (Tx Index 1)
...     {"Date": "2024-09-05", "Tx Index": 1, "Asset": "NVDA",
...      "Amount (asset)": -4.0, "Sell price ($)": 130.0,
...      "Buy price ($)": float('nan'), "Type": "Sell"},
...     {"Date": "2024-09-05", "Tx Index": 1, "Asset": "USD",
...      "Amount (asset)": 520.0, "Sell price ($)": float('nan'),
...      "Buy price ($)": 1.0, "Type": "Sell"},
...     {"Date": "2024-09-05", "Tx Index": 1, "Asset": "feeUSD",
...      "Amount (asset)": -1.0, "Sell price ($)": 1.0,
...      "Buy price ($)": float('nan'), "Type": "Sell"},
... ])
>>> rows = run_fifo_pipeline(df)
>>> len(rows) >= 1
True
>>> sorted(rows[0].keys())
['Cost Basis', 'Date Acquired', 'Date Sold', 'Description',
 'Gain or Loss', 'Proceeds']
>>> any(r["Description"].endswith("NVDA") for r in rows)
True
src.irs_asset_fifo_calculator.calculate_taxes.update_fifo(buy_data: AssetData, sell_data: AssetData, fee_data: AssetData, form8949: list[dict[str, str]], fifo: defaultdict[str, deque[FifoLot]]) → None

Updates FIFO dict of deque using info from this block of transactions.

Parameters:

  • buy_data (AssetData) – buy info for this block of transactions

  • sell_data (AssetData) – sell info for this block of transactions

  • fee_data (AssetData) – fee info for this block of transactions

  • form8949 (list[dict[str, str]]) – Form 8949 list of dicts holding txs.

  • fifo (defaultdict[str, deque[FifoLot]]) – purchases of each token defined by their amount, price, cost, and date

Returns:

None

Notes: - In general, buy and sell assets and fee asset should not be the same. If they were that way upstream, the fees should have already been added to buy or sell and then set to 0. - If previously calculated fees are same asset as buy and larger than buy amount, the net buy amount is negative and is thus reduced from FIFO instead of appended.

Example

Simple NVDA purchase that appends a new lot to the FIFO ledger: >>> from collections import defaultdict, deque >>> from datetime import date >>> from calculate_taxes import (AssetData, update_fifo) >>> fifo = defaultdict(deque) >>> form8949 = [] >>> buy = AssetData(asset=”NVDA”, amount=10.0, price=100.0, … total=1000.0, tx_date=date(2024, 1, 1)) >>> sell = AssetData(asset=”USD”, amount=-1000.0, price=1.0, … total=0.0, tx_date=date(2024, 1, 1)) >>> fee = AssetData(asset=None, amount=0.0, price=0.0, total=0.0, … tx_date=date(2024, 1, 1)) >>> update_fifo(buy, sell, fee, form8949, fifo) >>> len(fifo[“NVDA”]) 1 >>> fifo[“NVDA”][0][“amount”] 10.0 >>> fifo[“NVDA”][0][“price”] 100.0 >>> fifo[“NVDA”][0][“cost”] 1000.0 >>> fifo[“NVDA”][0][“tx_date”] datetime.date(2024, 1, 1) >>> form8949 []