diff --git a/pyproject.toml b/pyproject.toml index c1de816..60facd6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "kalshi-csv" -version = "0.1.2" +version = "0.2.0" description = "Parse Kalshi transaction CSV files and generate IRS Form 8949 tax summaries" readme = "README.md" license = "MIT" diff --git a/src/kalshi_csv/__init__.py b/src/kalshi_csv/__init__.py index 3f9929b..c0492ee 100644 --- a/src/kalshi_csv/__init__.py +++ b/src/kalshi_csv/__init__.py @@ -1,4 +1,4 @@ -__version__ = "0.1.2" +__version__ = "0.2.0" from .parser import KalshiCSV diff --git a/src/kalshi_csv/categories.py b/src/kalshi_csv/categories.py new file mode 100644 index 0000000..061b3ae --- /dev/null +++ b/src/kalshi_csv/categories.py @@ -0,0 +1,44 @@ +TICKER_CATEGORY_MAP = { + "KXMLB": "MLB Baseball", + "KXMLBHR": "MLB Baseball", + "KXMLBMEN": "MLB Baseball", + "KXNPB": "NPB Baseball (Japan)", + "KXNBASUMMER": "NBA Summer League", + "KXNEXTTEAMNBA": "NBA Summer League", + "KXWNBA": "WNBA Basketball", + "KXINXU": "S&P 500 (INXU Intraday)", + "KXINX": "S&P 500 (INXU Intraday)", + "KXMVESPORTS": "Esports & Gaming", +} + +SOCCER_PREFIXES = [ + "KXWC", + "KXUCL", + "KXUECL", + "KXBRASILEIRO", + "KXALLSVENSKAN", + "KXELITESERIEN", + "KXECULP", + "KXLIGAMX", + "KXLIGAEXP", + "KXKLEAGUE", + "KXCLUBF", + "KXSCOCUP", + "KXURYPD", + "KXDIMAYOR", + "KXARGPREM", + "KXBOLP", +] + + +def categorize_ticker(ticker): + """Maps a Kalshi market ticker to a human-readable category.""" + for prefix, category in TICKER_CATEGORY_MAP.items(): + if ticker.startswith(prefix): + return category + + for prefix in SOCCER_PREFIXES: + if ticker.startswith(prefix): + return "Global Soccer / Football" + + return "Other Markets" diff --git a/src/kalshi_csv/cli.py b/src/kalshi_csv/cli.py index 276ae3d..0466fdf 100644 --- a/src/kalshi_csv/cli.py +++ b/src/kalshi_csv/cli.py @@ -1,4 +1,5 @@ import argparse +import os import sys from .parser import KalshiCSV @@ -35,6 +36,17 @@ def main(): action="store_true", help="Use ASCII characters instead of Unicode box-drawing", ) + parser.add_argument( + "--legacy-web", + action="store_true", + help="Start a legacy web server (HTML 4.01) to view portfolio in browser", + ) + parser.add_argument( + "--legacy-web-port", + type=int, + default=8080, + help="Port for legacy web server (default: 8080)", + ) args = parser.parse_args() no_color = args.no_color @@ -43,6 +55,13 @@ def main(): kalshi = KalshiCSV(args.csv_path) kalshi.parse() + if args.legacy_web: + from .web import LegacyWebServer + csv_filename = os.path.basename(args.csv_path) + server = LegacyWebServer(kalshi, csv_filename, port=args.legacy_web_port) + server.serve() + return + headers = ["Ticker", "Side", "Qty", "Entry", "Exit", "P&L (No Fees)", "Fees"] widths = [32, 4, 6, 6, 6, 14, 6] diff --git a/src/kalshi_csv/parser.py b/src/kalshi_csv/parser.py index d072147..9f60c13 100644 --- a/src/kalshi_csv/parser.py +++ b/src/kalshi_csv/parser.py @@ -1,7 +1,10 @@ import csv import os +from collections import defaultdict from datetime import datetime +from .categories import categorize_ticker + class KalshiCSV: """Parses Kalshi transaction CSV data and calculates tax-relevant aggregates.""" @@ -18,6 +21,11 @@ class KalshiCSV: "total_tax_proceeds": 0.0, "earliest_open_date": None, "latest_close_date": None, + "wins": 0, + "losses": 0, + "pushes": 0, + "best_trade": None, + "worst_trade": None, } def parse(self): @@ -40,8 +48,22 @@ class KalshiCSV: open_fees = float(row["open_fees_dollars"]) close_fees = float(row["close_fees_dollars"]) + open_dt = None + close_dt = None + if row.get("open_timestamp"): + try: + open_dt = datetime.fromisoformat(row["open_timestamp"]) + except ValueError: + pass + if row.get("close_timestamp"): + try: + close_dt = datetime.fromisoformat(row["close_timestamp"]) + except ValueError: + pass + + ticker = row["market_ticker"] trade = { - "ticker": row["market_ticker"], + "ticker": ticker, "side": row["side"].upper(), "qty": qty, "entry": entry, @@ -50,6 +72,9 @@ class KalshiCSV: "pnl_with_fees": pnl_with_fees, "open_fees": open_fees, "close_fees": close_fees, + "open_timestamp": open_dt, + "close_timestamp": close_dt, + "market_category": categorize_ticker(ticker), } self.trades.append(trade) @@ -60,27 +85,37 @@ class KalshiCSV: self.summary["total_pnl_with_fees"] += pnl_with_fees self.summary["total_fees"] += open_fees + close_fees - if row.get("open_timestamp"): - try: - open_dt = datetime.fromisoformat(row["open_timestamp"]) - if ( - self.summary["earliest_open_date"] is None - or open_dt < self.summary["earliest_open_date"] - ): - self.summary["earliest_open_date"] = open_dt - except ValueError: - pass + if pnl_with_fees > 0: + self.summary["wins"] += 1 + elif pnl_with_fees < 0: + self.summary["losses"] += 1 + else: + self.summary["pushes"] += 1 - if row.get("close_timestamp"): - try: - close_dt = datetime.fromisoformat(row["close_timestamp"]) - if ( - self.summary["latest_close_date"] is None - or close_dt > self.summary["latest_close_date"] - ): - self.summary["latest_close_date"] = close_dt - except ValueError: - pass + if ( + self.summary["best_trade"] is None + or pnl_with_fees > self.summary["best_trade"]["pnl_with_fees"] + ): + self.summary["best_trade"] = trade + if ( + self.summary["worst_trade"] is None + or pnl_with_fees < self.summary["worst_trade"]["pnl_with_fees"] + ): + self.summary["worst_trade"] = trade + + if open_dt is not None: + if ( + self.summary["earliest_open_date"] is None + or open_dt < self.summary["earliest_open_date"] + ): + self.summary["earliest_open_date"] = open_dt + + if close_dt is not None: + if ( + self.summary["latest_close_date"] is None + or close_dt > self.summary["latest_close_date"] + ): + self.summary["latest_close_date"] = close_dt return self @@ -101,3 +136,36 @@ class KalshiCSV: "cost_basis": self.summary["total_tax_basis"], "gain_or_loss": self.summary["total_pnl_with_fees"], } + + def market_breakdown(self): + """Returns a list of dicts with market category breakdown sorted by trade count.""" + categories = defaultdict(lambda: {"trades": 0, "wins": 0, "net_pnl": 0.0}) + + for trade in self.trades: + cat = trade["market_category"] + categories[cat]["trades"] += 1 + categories[cat]["net_pnl"] += trade["pnl_with_fees"] + if trade["pnl_with_fees"] > 0: + categories[cat]["wins"] += 1 + + breakdown = [] + for cat, data in categories.items(): + win_rate = (data["wins"] / data["trades"] * 100) if data["trades"] > 0 else 0 + breakdown.append({ + "category": cat, + "trades": data["trades"], + "win_rate": win_rate, + "net_pnl": data["net_pnl"], + }) + + return sorted(breakdown, key=lambda x: x["trades"], reverse=True) + + def recent_closed_positions(self, n=20): + """Returns the last n trades sorted by close_timestamp descending.""" + trades_with_close = [t for t in self.trades if t["close_timestamp"] is not None] + sorted_trades = sorted( + trades_with_close, + key=lambda t: t["close_timestamp"], + reverse=True, + ) + return sorted_trades[:n] diff --git a/src/kalshi_csv/web.py b/src/kalshi_csv/web.py new file mode 100644 index 0000000..4e12aca --- /dev/null +++ b/src/kalshi_csv/web.py @@ -0,0 +1,239 @@ +import html +from datetime import datetime +from http.server import HTTPServer, BaseHTTPRequestHandler + +from . import __version__ + + +def render_portfolio_html(kalshi, csv_filename): + """Renders the full HTML 4.01 portfolio page from parsed Kalshi data.""" + summary = kalshi.summary + market_breakdown = kalshi.market_breakdown() + recent_positions = kalshi.recent_closed_positions(20) + + period_end = summary["latest_close_date"] + if period_end: + period_end_str = period_end.strftime("%B %d, %Y").upper() + else: + period_end_str = "N/A" + + net_pnl = summary["total_pnl_with_fees"] + net_pnl_color = "#006600" if net_pnl >= 0 else "#990000" + net_pnl_str = f"${net_pnl:+.2f}" + + wins = summary["wins"] + losses = summary["losses"] + pushes = summary["pushes"] + total = summary["trade_count"] + win_pct = (wins / total * 100) if total > 0 else 0 + push_note = f" ({pushes} Push{'s' if pushes != 1 else ''})" if pushes > 0 else "" + + best_trade = summary["best_trade"] + worst_trade = summary["worst_trade"] + if best_trade and worst_trade: + best_pnl = best_trade["pnl_with_fees"] + worst_pnl = worst_trade["pnl_with_fees"] + best_cat = best_trade["market_category"] + worst_cat = worst_trade["market_category"] + best_color = "#006600" if best_pnl >= 0 else "#990000" + worst_color = "#006600" if worst_pnl >= 0 else "#990000" + best_str = f"${best_pnl:+.2f}" + worst_str = f"${worst_pnl:+.2f}" + best_worst_subtext = f"{best_cat} / {worst_cat}" + else: + best_str = "$0.00" + worst_str = "$0.00" + best_color = "#006600" + worst_color = "#990000" + best_worst_subtext = "N/A" + + rows_html = "" + for i, item in enumerate(market_breakdown): + pnl = item["net_pnl"] + pnl_color = "#006600" if pnl >= 0 else "#990000" + pnl_str = f"${pnl:+.2f}" + win_rate_str = f"{item['win_rate']:.1f}%" + rows_html += f""" + + {html.escape(item['category'])} + {item['trades']} + {win_rate_str} + {pnl_str} + """ + if i < len(market_breakdown) - 1: + rows_html += """ +
""" + + positions_html = "" + for trade in recent_positions: + close_dt = trade["close_timestamp"] + date_str = close_dt.strftime("%m/%d %H:%M") if close_dt else "N/A" + ticker = html.escape(trade["ticker"]) + side = trade["side"] + qty = f"{trade['qty']:.2f}" + entry = f"${trade['entry']:.2f}" + exit_val = f"${trade['exit']:.2f}" + pnl = trade["pnl_with_fees"] + pnl_color = "#006600" if pnl >= 0 else "#990000" + pnl_str = f"${pnl:+.2f}" + positions_html += f""" + + {date_str} + {ticker} + {side} + {qty} + {entry} + {exit_val} + {pnl_str} + """ + + page = f""" + + + Kalshi Portfolio Statement + + + + +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+ KALSHI DERIVATIVES / ACCOUNT AUDIT
+ Year-End Performance Summary
+ PERIOD ENDING: {html.escape(period_end_str)}  |  SOURCE: {html.escape(csv_filename)} +
+
+
+ + + + + + + +
+ NET REALIZED P&L
+ {net_pnl_str}
+ Includes ${summary['total_fees']:.2f} fees + 		+ WIN / LOSS RECORD
+ {wins} - {losses}
+ {pushes} Push{'s' if pushes != 1 else ''} ({win_pct:.1f}% Win) + 		+ TOTAL VOLUME
+ {total}
+ Executed Contracts + 		+ BEST/WORST SINGLE
+ {best_str} / {worst_str}
+ {html.escape(best_worst_subtext)} +
+


+ Market Breakdown +

+ + + + + + + + + {rows_html} +
ASSET CLASS / MARKETTRADESWIN RATENET P&L
+


+ Recent Closed Positions +

+ + + + + + + + + + + + {positions_html} +
DATE/TIMETICKERSIDEQTYENTRYEXITP&L
+

+ + {html.escape(csv_filename)} • rendered with vanilla HTML 4.01 strict table markup • kalshi-csv v{__version__} + +
+
+ + +""" + return page + + +class LegacyWebHandler(BaseHTTPRequestHandler): + """HTTP request handler that serves the legacy portfolio page.""" + + def do_GET(self): + if self.path == "/" or self.path == "/index.html": + self.send_response(200) + self.send_header("Content-type", "text/html; charset=iso-8859-1") + self.end_headers() + html_content = self.server.html_content + self.wfile.write(html_content.encode("iso-8859-1")) + else: + self.send_error(404, "Not Found") + + def log_message(self, format, *args): + pass + + +class LegacyWebServer: + """HTTP server for the legacy portfolio view.""" + + def __init__(self, kalshi, csv_filename, host="0.0.0.0", port=8080): + self.kalshi = kalshi + self.csv_filename = csv_filename + self.host = host + self.port = port + self.html_content = render_portfolio_html(kalshi, csv_filename) + + def serve(self): + """Starts the HTTP server and blocks until interrupted.""" + server = HTTPServer((self.host, self.port), LegacyWebHandler) + server.html_content = self.html_content + print(f"Serving legacy portfolio view at http://{self.host}:{self.port}/") + print("Press Ctrl+C to stop.") + try: + server.serve_forever() + except KeyboardInterrupt: + print("\nShutting down server.") + server.server_close() diff --git a/tests/test_categories.py b/tests/test_categories.py new file mode 100644 index 0000000..644ed71 --- /dev/null +++ b/tests/test_categories.py @@ -0,0 +1,47 @@ +from kalshi_csv.categories import categorize_ticker + + +def test_mlb_categorization(): + assert categorize_ticker("KXMLBGAME-26JUL081940BOSCWS-BOS") == "MLB Baseball" + assert categorize_ticker("KXMLBHRDERBY-26-KSCHWARBER12") == "MLB Baseball" + + +def test_npb_categorization(): + assert categorize_ticker("KXNPBGAME-26JUL150500YOMYAK-YAK") == "NPB Baseball (Japan)" + + +def test_nba_summer_categorization(): + assert categorize_ticker("KXNBASUMMERGAME-26JUL14MEMGSW-GSW") == "NBA Summer League" + + +def test_wnba_categorization(): + assert categorize_ticker("KXWNBAGAME-26JUL13PHXMIN-PHX") == "WNBA Basketball" + + +def test_sp500_categorization(): + assert categorize_ticker("KXINXU-26JUL08H1400-T7479.9999") == "S&P 500 (INXU Intraday)" + assert categorize_ticker("KXINX-26JUL08H1400-T7479.9999") == "S&P 500 (INXU Intraday)" + + +def test_esports_categorization(): + assert categorize_ticker("KXMVESPORTSMULTIGAMEEXTENDED-S2026769CE3FA3F9-6D4DB2E2128") == "Esports & Gaming" + + +def test_soccer_categorization(): + assert categorize_ticker("KXWCADVANCE-26JUL07ARGEGY-ARG") == "Global Soccer / Football" + assert categorize_ticker("KXUCLADVANCE-26JUL14KUPSVAR-VAR") == "Global Soccer / Football" + assert categorize_ticker("KXBRASILEIROBGAME-26JUL13AMGLON-LON") == "Global Soccer / Football" + assert categorize_ticker("KXECULPGAME-26JUL14MACMUR-MUR") == "Global Soccer / Football" + assert categorize_ticker("KXALLSVENSKANGAME-26JUL12BROSIR-SIR") == "Global Soccer / Football" + assert categorize_ticker("KXCLUBFGAME-26JUL27GALVEN-VEN") == "Global Soccer / Football" + + +def test_other_markets_categorization(): + assert categorize_ticker("KXRAIN-26JUL15-ATL") == "Other Markets" + assert categorize_ticker("KXTRUMPMENTION-26JUL15") == "Other Markets" + assert categorize_ticker("KXTEMPNYCH-26JUL15") == "Other Markets" + assert categorize_ticker("KXHIGHCHI-26JUL15") == "Other Markets" + + +def test_unknown_ticker_defaults_to_other(): + assert categorize_ticker("UNKNOWN-TICKER-123") == "Other Markets" diff --git a/tests/test_cli.py b/tests/test_cli.py index 13ac9ac..5e67d76 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -103,3 +103,24 @@ def test_cli_ascii_flag(sample_csv): assert "|" in result.stdout assert "┌" not in result.stdout assert "│" not in result.stdout + + +def test_cli_legacy_web_flag_in_help(): + result = subprocess.run( + [sys.executable, "-m", "kalshi_csv.cli", "--help"], + capture_output=True, + text=True, + ) + assert result.returncode == 0 + assert "--legacy-web" in result.stdout + assert "--legacy-web-port" in result.stdout + + +def test_cli_legacy_web_port_default_in_help(): + result = subprocess.run( + [sys.executable, "-m", "kalshi_csv.cli", "--help"], + capture_output=True, + text=True, + ) + assert result.returncode == 0 + assert "8080" in result.stdout diff --git a/tests/test_parser.py b/tests/test_parser.py index 83a77a6..f35eb92 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -89,3 +89,88 @@ def test_file_not_found(): kalshi = KalshiCSV("/nonexistent/path.csv") with pytest.raises(FileNotFoundError): kalshi.parse() + + +def test_market_breakdown_returns_list(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + breakdown = kalshi.market_breakdown() + assert isinstance(breakdown, list) + assert len(breakdown) > 0 + + +def test_market_breakdown_structure(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + breakdown = kalshi.market_breakdown() + for item in breakdown: + assert "category" in item + assert "trades" in item + assert "win_rate" in item + assert "net_pnl" in item + + +def test_market_breakdown_sorted_by_trades(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + breakdown = kalshi.market_breakdown() + trade_counts = [item["trades"] for item in breakdown] + assert trade_counts == sorted(trade_counts, reverse=True) + + +def test_recent_closed_positions_returns_list(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + positions = kalshi.recent_closed_positions() + assert isinstance(positions, list) + + +def test_recent_closed_positions_limit(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + positions = kalshi.recent_closed_positions(n=2) + assert len(positions) <= 2 + + +def test_recent_closed_positions_sorted_by_date(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + positions = kalshi.recent_closed_positions() + if len(positions) > 1: + timestamps = [p["close_timestamp"] for p in positions] + assert timestamps == sorted(timestamps, reverse=True) + + +def test_summary_tracks_wins_losses_pushes(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + assert kalshi.summary["wins"] >= 0 + assert kalshi.summary["losses"] >= 0 + assert kalshi.summary["pushes"] >= 0 + assert kalshi.summary["wins"] + kalshi.summary["losses"] + kalshi.summary["pushes"] == kalshi.summary["trade_count"] + + +def test_summary_tracks_best_worst_trade(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + assert kalshi.summary["best_trade"] is not None + assert kalshi.summary["worst_trade"] is not None + assert "pnl_with_fees" in kalshi.summary["best_trade"] + assert "pnl_with_fees" in kalshi.summary["worst_trade"] + assert kalshi.summary["best_trade"]["pnl_with_fees"] >= kalshi.summary["worst_trade"]["pnl_with_fees"] + + +def test_trade_has_market_category(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + for trade in kalshi.trades: + assert "market_category" in trade + assert isinstance(trade["market_category"], str) + + +def test_trade_has_timestamps(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + for trade in kalshi.trades: + assert "open_timestamp" in trade + assert "close_timestamp" in trade diff --git a/tests/test_web.py b/tests/test_web.py new file mode 100644 index 0000000..d8779ba --- /dev/null +++ b/tests/test_web.py @@ -0,0 +1,96 @@ +from kalshi_csv import KalshiCSV +from kalshi_csv.web import render_portfolio_html + + +def test_render_html_contains_header(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + html_content = render_portfolio_html(kalshi, "test.csv") + assert "KALSHI DERIVATIVES / ACCOUNT AUDIT" in html_content + assert "Year-End Performance Summary" in html_content + + +def test_render_html_contains_summary_metrics(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + html_content = render_portfolio_html(kalshi, "test.csv") + assert "NET REALIZED P&L" in html_content + assert "WIN / LOSS RECORD" in html_content + assert "TOTAL VOLUME" in html_content + assert "BEST/WORST SINGLE" in html_content + + +def test_render_html_contains_market_breakdown(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + html_content = render_portfolio_html(kalshi, "test.csv") + assert "Market Breakdown" in html_content + assert "ASSET CLASS / MARKET" in html_content + assert "TRADES" in html_content + assert "WIN RATE" in html_content + + +def test_render_html_contains_recent_positions(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + html_content = render_portfolio_html(kalshi, "test.csv") + assert "Recent Closed Positions" in html_content + assert "DATE/TIME" in html_content + assert "TICKER" in html_content + assert "SIDE" in html_content + assert "ENTRY" in html_content + assert "EXIT" in html_content + + +def test_render_html_contains_trade_data(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + html_content = render_portfolio_html(kalshi, "test.csv") + assert "TESTMARKET-WIN" in html_content + assert "TESTMARKET-LOSS" in html_content + assert "TESTMARKET-SMALL" in html_content + + +def test_render_html_html401_doctype(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + html_content = render_portfolio_html(kalshi, "test.csv") + assert '3<" in html_content + + +def test_render_html_shows_csv_filename(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + html_content = render_portfolio_html(kalshi, "my-kalshi-data.csv") + assert "my-kalshi-data.csv" in html_content + + +def test_render_html_escapes_html_in_tickers(sample_csv): + kalshi = KalshiCSV(sample_csv) + kalshi.parse() + kalshi.trades[0]["ticker"] = "" + html_content = render_portfolio_html(kalshi, "test.csv") + assert "" not in html_content + assert "<script>" in html_content