markmental/kalshi-csv
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Structure into library for pip

Browse source
This commit is contained in:
mark 2026-07-18 13:34:47 -04:00
commit 25ad16dac2
12 changed files with 540 additions and 99 deletions
Download patch file Download diff file Expand all files Collapse all files

49
.gitignore vendored
View file

@ -1,2 +1,51 @@
# Private Kalshi csvs
*.csv
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
*.so
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual environments
.venv/
venv/
ENV/
# Pytest
.pytest_cache/
.coverage
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# Jupyter Notebook
.ipynb_checkpoints
# Environment variables
.env

21
LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 Mark Robillard Jr
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

88
README.md Normal file
View file

@ -0,0 +1,88 @@
# kalshi-csv
Parse Kalshi transaction CSV files and generate IRS Form 8949 tax summaries for event contract trading.
## Installation
```bash
pip install kalshi-csv
```
## CLI Usage
Parse a Kalshi transactions CSV and display the trade matrix with IRS summary:
```bash
kalshi-csv Kalshi-Transactions-2026.csv
```
Export the IRS summary to a file:
```bash
kalshi-csv Kalshi-Transactions-2026.csv --irs-file irs-summary.txt
```
Disable colored output (useful for piping or redirecting):
```bash
kalshi-csv Kalshi-Transactions-2026.csv --no-color
```
### Sample Output
```
Ticker | Side | Qty | Entry | Exit | P&L (No Fees)
-------------------------------------------------------------------------------
KXWCADVANCE-26JUL07ARGEGY-ARG | YES | 0.17 | $0.86 | $0.69 | -$0.03
KXWC1H-26JUL07ARGEGY-TIE | YES | 0.34 | $0.28 | $0.00 | -$0.10
KXMLBGAME-26JUL081940BOSCWS-BOS | YES | 0.96 | $0.50 | $0.91 | $0.39
-------------------------------------------------------------------------------
Total Transactions Parsed: 3
Total Exchange Fees Paid: $0.05
Internal Tracked Net P&L: $0.26
-------------------------------------------------------------------------------
=== IRS FORM 8949 / SCHEDULE D AGGREGATE SUMMARY ===
Use these exact aggregates for a single-line summary entry:
* Box to Check: Box C (Short-term, not reported on Form 1099-B)
* (a) Description: Kalshi Event Contracts (Aggregate Summary)
* (d) Gross Proceeds: $2.50
* (e) Cost or Other Basis: $2.24
* (h) Gain or (Loss): $0.26
====================================================
```
## Library API
Use `kalshi-csv` as a Python library in your own scripts:
```python
from kalshi_csv import KalshiCSV
kalshi = KalshiCSV("Kalshi-Transactions-2026.csv")
kalshi.parse()
print(f"Total trades: {kalshi.summary['trade_count']}")
print(f"Total P&L: ${kalshi.summary['total_pnl_with_fees']:.2f}")
irs = kalshi.irs_summary()
print(f"Gross Proceeds: ${irs['gross_proceeds']:.2f}")
print(f"Cost Basis: ${irs['cost_basis']:.2f}")
print(f"Gain/Loss: ${irs['gain_or_loss']:.2f}")
```
## IRS Form 8949
Kalshi event contracts are typically reported on **IRS Form 8949, Box C** (short-term transactions not reported on Form 1099-B). The tool calculates:
- **Gross Proceeds**: Total exit value minus close fees
- **Cost Basis**: Total entry value plus open fees
- **Gain/Loss**: Realized P&L including all fees
Use the aggregate summary for a single-line entry on Form 8949, or export to a file for your records.
**Disclaimer**: This tool provides calculations based on Kalshi transaction data. Consult a tax professional for specific tax advice.
## License
MIT

99
kalshi-parse.py
View file

@ -1,99 +0,0 @@
#!/usr/bin/env python3
import csv
import sys
import os
class KalshiCSV:
"""Parses Kalshi transaction CSV data and handles terminal color-coding
and tax summary generation.
"""
def __init__(self, file_path):
self.file_path = file_path
self.trade_count = 0
self.total_tax_basis = 0.0
self.total_tax_proceeds = 0.0
self.total_pnl_without_fees = 0.0
self.total_pnl_with_fees = 0.0
self.total_fees = 0.0
def parse(self):
"""Processes the CSV file row-by-row and prints the live trade matrix."""
if not os.path.exists(self.file_path):
print(f"Error: File '{self.file_path}' not found.")
sys.exit(1)
print(f"\n{'Ticker':<32} | {'Side':<4} | {'Qty':<6} | {'Entry':<6} | {'Exit':<6} | {'P&L (No Fees)':<14}")
print("-" * 83)
with open(self.file_path, mode='r', newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
if not row.get('realized_pnl_without_fees_dollars'):
continue
ticker = row['market_ticker']
side = row['side'].upper()
qty = float(row['quantity_fp'])
entry = float(row['entry_price_dollars'])
exit_val = float(row['exit_price_dollars'])
pnl_no_fees = float(row['realized_pnl_without_fees_dollars'])
pnl_with_fees = float(row['realized_pnl_with_fees_dollars'])
open_fees = float(row['open_fees_dollars'])
close_fees = float(row['close_fees_dollars'])
# --- Aggregate Calculations ---
self.total_tax_basis += (qty * entry) + open_fees
self.total_tax_proceeds += (qty * exit_val) - close_fees
self.total_pnl_without_fees += pnl_no_fees
self.total_pnl_with_fees += pnl_with_fees
self.total_fees += (open_fees + close_fees)
self.trade_count += 1
# Format row P&L using the color methods
pnl_str = self.format_currency_color(pnl_no_fees)
print(f"{ticker:<32} | {side:<4} | {qty:<6.2f} | ${entry:<5.2f} | ${exit_val:<5.2f} | {pnl_str:<14}")
print("-" * 83)
print(f"Total Transactions Parsed: {self.trade_count}")
print(f"Total Exchange Fees Paid: ${self.total_fees:.2f}")
print(f"Internal Tracked Net P&L: " + self.format_currency_color(self.total_pnl_without_fees))
print("-" * 83)
def color_green(self, text):
"""Wraps text in ANSI green."""
return f"\033[1;32m{text}\033[0m"
def color_red(self, text):
"""Wraps text in ANSI red."""
return f"\033[1;31m{text}\033[0m"
def format_currency_color(self, value):
"""Returns a signed, colorized string based on profit or loss status."""
val_str = f"${value:+.2f}"
return self.color_green(val_str) if value >= 0 else self.color_red(val_str)
def summarizeIRS(self):
"""Outputs the structured block needed for a single-line entry on IRS Form 8949."""
print("\033[1;33m=== IRS FORM 8949 / SCHEDULE D AGGREGATE SUMMARY ===\033[0m")
print("Use these exact aggregates for a single-line summary entry:")
print(f" * Box to Check: \033[1;37mBox C\033[0m (Short-term, not reported on Form 1099-B)")
print(f" * (a) Description: Kalshi Event Contracts (Aggregate Summary)")
print(f" * (d) Gross Proceeds: " + self.color_green(f"${self.total_tax_proceeds:.2f}"))
print(f" * (e) Cost or Other Basis: \033[1;36m${self.total_tax_basis:.2f}\033[0m")
print(f" * (h) Gain or (Loss): " + self.format_currency_color(self.total_pnl_with_fees))
print("\033[1;33m====================================================\033[0m\n")
if __name__ == '__main__':
if len(sys.argv) < 2:
print("Usage: python3 kalshi_parser.py <path_to_transactions_csv>")
sys.exit(1)
# Instantiate, parse the matrix, and drop the IRS aggregate summary
parser = KalshiCSV(sys.argv[1])
parser.parse()
parser.summarizeIRS()

33
pyproject.toml Normal file
View file

@ -0,0 +1,33 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "kalshi-csv"
version = "0.1.0"
description = "Parse Kalshi transaction CSV files and generate IRS Form 8949 tax summaries"
readme = "README.md"
license = "MIT"
requires-python = ">=3.8"
authors = [
{ name = "markmental", email = "marky611@gmail.com" }
]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: End Users/Desktop",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Office/Business :: Financial",
]
keywords = ["kalshi", "csv", "tax", "irs", "form-8949"]
[project.scripts]
kalshi-csv = "kalshi_csv.cli:main"
[tool.hatch.build.targets.wheel]
packages = ["src/kalshi_csv"]

5
src/kalshi_csv/__init__.py Normal file
View file

@ -0,0 +1,5 @@
__version__ = "0.1.0"
from .parser import KalshiCSV
__all__ = ["KalshiCSV", "__version__"]

80
src/kalshi_csv/cli.py Normal file
View file

@ -0,0 +1,80 @@
import argparse
import sys
from .parser import KalshiCSV
from .formatter import (
color_white,
color_yellow,
color_cyan,
format_currency_color,
)
def main():
parser = argparse.ArgumentParser(
description="Parse Kalshi transaction CSV and generate IRS tax summary."
)
parser.add_argument("csv_path", help="Path to Kalshi transactions CSV file")
parser.add_argument(
"--irs-file",
help="Write IRS Form 8949 summary to this file",
)
parser.add_argument(
"--no-color",
action="store_true",
help="Disable ANSI color output",
)
args = parser.parse_args()
no_color = args.no_color
kalshi = KalshiCSV(args.csv_path)
kalshi.parse()
print()
print(
f"{'Ticker':<32} | {'Side':<4} | {'Qty':<6} | {'Entry':<6} | {'Exit':<6} | {'P&L (No Fees)':<14}"
)
print("-" * 83)
for trade in kalshi.trades:
pnl_str = format_currency_color(trade["pnl_no_fees"], no_color)
print(
f"{trade['ticker']:<32} | {trade['side']:<4} | {trade['qty']:<6.2f} | "
f"${trade['entry']:<5.2f} | ${trade['exit']:<5.2f} | {pnl_str:<14}"
)
print("-" * 83)
print(f"Total Transactions Parsed: {kalshi.summary['trade_count']}")
print(f"Total Exchange Fees Paid: ${kalshi.summary['total_fees']:.2f}")
print(
f"Internal Tracked Net P&L: "
+ format_currency_color(kalshi.summary["total_pnl_without_fees"], no_color)
)
print("-" * 83)
irs = kalshi.irs_summary()
print(color_yellow("=== IRS FORM 8949 / SCHEDULE D AGGREGATE SUMMARY ===", no_color))
print("Use these exact aggregates for a single-line summary entry:")
print(f" * Box to Check: {color_white('Box C', no_color)} (Short-term, not reported on Form 1099-B)")
print(f" * (a) Description: {irs['description']}")
print(f" * (d) Gross Proceeds: {format_currency_color(irs['gross_proceeds'], no_color)}")
print(f" * (e) Cost or Other Basis: {color_cyan(f'${irs['cost_basis']:.2f}', no_color)}")
print(f" * (h) Gain or (Loss): {format_currency_color(irs['gain_or_loss'], no_color)}")
print(color_yellow("====================================================", no_color))
print()
if args.irs_file:
with open(args.irs_file, "w", encoding="utf-8") as f:
f.write("IRS FORM 8949 / SCHEDULE D AGGREGATE SUMMARY\n")
f.write("Use these exact aggregates for a single-line summary entry:\n")
f.write(f" * Box to Check: Box C (Short-term, not reported on Form 1099-B)\n")
f.write(f" * (a) Description: {irs['description']}\n")
f.write(f" * (d) Gross Proceeds: ${irs['gross_proceeds']:.2f}\n")
f.write(f" * (e) Cost or Other Basis: ${irs['cost_basis']:.2f}\n")
f.write(f" * (h) Gain or (Loss): ${irs['gain_or_loss']:.2f}\n")
print(f"IRS summary written to: {args.irs_file}")
if __name__ == "__main__":
main()

39
src/kalshi_csv/formatter.py Normal file
View file

@ -0,0 +1,39 @@
def color_green(text, no_color=False):
"""Wraps text in ANSI green."""
if no_color:
return text
return f"\033[1;32m{text}\033[0m"
def color_red(text, no_color=False):
"""Wraps text in ANSI red."""
if no_color:
return text
return f"\033[1;31m{text}\033[0m"
def color_yellow(text, no_color=False):
"""Wraps text in ANSI yellow."""
if no_color:
return text
return f"\033[1;33m{text}\033[0m"
def color_cyan(text, no_color=False):
"""Wraps text in ANSI cyan."""
if no_color:
return text
return f"\033[1;36m{text}\033[0m"
def color_white(text, no_color=False):
"""Wraps text in ANSI white."""
if no_color:
return text
return f"\033[1;37m{text}\033[0m"
def format_currency_color(value, no_color=False):
"""Returns a signed, colorized string based on profit or loss status."""
val_str = f"${value:+.2f}"
return color_green(val_str, no_color) if value >= 0 else color_red(val_str, no_color)

70
src/kalshi_csv/parser.py Normal file
View file

@ -0,0 +1,70 @@
import csv
import os
class KalshiCSV:
"""Parses Kalshi transaction CSV data and calculates tax-relevant aggregates."""
def __init__(self, file_path):
self.file_path = file_path
self.trades = []
self.summary = {
"trade_count": 0,
"total_fees": 0.0,
"total_pnl_without_fees": 0.0,
"total_pnl_with_fees": 0.0,
"total_tax_basis": 0.0,
"total_tax_proceeds": 0.0,
}
def parse(self):
"""Processes the CSV file row-by-row and populates trades and summary."""
if not os.path.exists(self.file_path):
raise FileNotFoundError(f"File '{self.file_path}' not found.")
with open(self.file_path, mode="r", newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
if not row.get("realized_pnl_without_fees_dollars"):
continue
qty = float(row["quantity_fp"])
entry = float(row["entry_price_dollars"])
exit_val = float(row["exit_price_dollars"])
pnl_no_fees = float(row["realized_pnl_without_fees_dollars"])
pnl_with_fees = float(row["realized_pnl_with_fees_dollars"])
open_fees = float(row["open_fees_dollars"])
close_fees = float(row["close_fees_dollars"])
trade = {
"ticker": row["market_ticker"],
"side": row["side"].upper(),
"qty": qty,
"entry": entry,
"exit": exit_val,
"pnl_no_fees": pnl_no_fees,
"pnl_with_fees": pnl_with_fees,
"open_fees": open_fees,
"close_fees": close_fees,
}
self.trades.append(trade)
self.summary["trade_count"] += 1
self.summary["total_tax_basis"] += (qty * entry) + open_fees
self.summary["total_tax_proceeds"] += (qty * exit_val) - close_fees
self.summary["total_pnl_without_fees"] += pnl_no_fees
self.summary["total_pnl_with_fees"] += pnl_with_fees
self.summary["total_fees"] += open_fees + close_fees
return self
def irs_summary(self):
"""Returns a dict with IRS Form 8949 aggregate fields."""
return {
"box": "C",
"description": "Kalshi Event Contracts (Aggregate Summary)",
"gross_proceeds": self.summary["total_tax_proceeds"],
"cost_basis": self.summary["total_tax_basis"],
"gain_or_loss": self.summary["total_pnl_with_fees"],
}

8
tests/conftest.py Normal file
View file

@ -0,0 +1,8 @@
import pytest
from pathlib import Path
@pytest.fixture
def sample_csv():
"""Returns the path to the synthetic test CSV fixture."""
return str(Path(__file__).parent / "fixtures" / "sample.csv")

67
tests/test_cli.py Normal file
View file

@ -0,0 +1,67 @@
import subprocess
import sys
from pathlib import Path
def test_cli_runs_successfully(sample_csv):
result = subprocess.run(
[sys.executable, "-m", "kalshi_csv.cli", sample_csv],
capture_output=True,
text=True,
)
assert result.returncode == 0
assert "TESTMARKET-WIN" in result.stdout
assert "TESTMARKET-LOSS" in result.stdout
assert "Total Transactions Parsed: 3" in result.stdout
def test_cli_irs_summary_output(sample_csv):
result = subprocess.run(
[sys.executable, "-m", "kalshi_csv.cli", sample_csv],
capture_output=True,
text=True,
)
assert "IRS FORM 8949" in result.stdout
assert "Box C" in result.stdout
assert "Gross Proceeds:" in result.stdout
assert "Cost or Other Basis:" in result.stdout
assert "Gain or (Loss):" in result.stdout
def test_cli_irs_file_creation(sample_csv, tmp_path):
irs_file = tmp_path / "irs-output.txt"
result = subprocess.run(
[sys.executable, "-m", "kalshi_csv.cli", sample_csv, "--irs-file", str(irs_file)],
capture_output=True,
text=True,
)
assert result.returncode == 0
assert irs_file.exists()
content = irs_file.read_text()
assert "IRS FORM 8949" in content
assert "Box C" in content
assert "Gross Proceeds:" in content
assert "Cost or Other Basis:" in content
assert "Gain or (Loss):" in content
assert "\033[" not in content
def test_cli_no_color_flag(sample_csv):
result = subprocess.run(
[sys.executable, "-m", "kalshi_csv.cli", sample_csv, "--no-color"],
capture_output=True,
text=True,
)
assert result.returncode == 0
assert "\033[" not in result.stdout
def test_cli_missing_file():
result = subprocess.run(
[sys.executable, "-m", "kalshi_csv.cli", "/nonexistent/file.csv"],
capture_output=True,
text=True,
)
assert result.returncode != 0
assert "not found" in result.stderr.lower() or "error" in result.stderr.lower()

80
tests/test_parser.py Normal file
View file

@ -0,0 +1,80 @@
import pytest
from kalshi_csv import KalshiCSV
def test_parse_returns_self(sample_csv):
kalshi = KalshiCSV(sample_csv)
result = kalshi.parse()
assert result is kalshi
def test_trade_count(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
assert kalshi.summary["trade_count"] == 3
def test_total_fees(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
assert abs(kalshi.summary["total_fees"] - 0.07) < 1e-6
def test_total_pnl_without_fees(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
assert abs(kalshi.summary["total_pnl_without_fees"] - (-0.20)) < 1e-6
def test_total_pnl_with_fees(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
assert abs(kalshi.summary["total_pnl_with_fees"] - (-0.27)) < 1e-6
def test_total_tax_basis(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
assert abs(kalshi.summary["total_tax_basis"] - 1.64) < 1e-6
def test_total_tax_proceeds(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
assert abs(kalshi.summary["total_tax_proceeds"] - 1.37) < 1e-6
def test_trades_list_length(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
assert len(kalshi.trades) == 3
def test_first_trade_data(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
trade = kalshi.trades[0]
assert trade["ticker"] == "TESTMARKET-WIN"
assert trade["side"] == "YES"
assert trade["qty"] == 1.0
assert trade["entry"] == 0.50
assert trade["exit"] == 1.00
assert abs(trade["pnl_no_fees"] - 0.50) < 1e-6
assert abs(trade["pnl_with_fees"] - 0.47) < 1e-6
def test_irs_summary(sample_csv):
kalshi = KalshiCSV(sample_csv)
kalshi.parse()
irs = kalshi.irs_summary()
assert irs["box"] == "C"
assert irs["description"] == "Kalshi Event Contracts (Aggregate Summary)"
assert abs(irs["gross_proceeds"] - 1.37) < 1e-6
assert abs(irs["cost_basis"] - 1.64) < 1e-6
assert abs(irs["gain_or_loss"] - (-0.27)) < 1e-6
def test_file_not_found():
kalshi = KalshiCSV("/nonexistent/path.csv")
with pytest.raises(FileNotFoundError):
kalshi.parse()