1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.10"
4# dependencies = ["yfinance>=0.2.40"]
5# ///
6"""
7Portfolio management for stock-analysis skill.
8 
9Usage:
10    uv run portfolio.py create "Portfolio Name"
11    uv run portfolio.py list
12    uv run portfolio.py show [--portfolio NAME]
13    uv run portfolio.py delete "Portfolio Name"
14    uv run portfolio.py rename "Old Name" "New Name"
15 
16    uv run portfolio.py add TICKER --quantity 100 --cost 150.00 [--portfolio NAME]
17    uv run portfolio.py update TICKER --quantity 150 [--portfolio NAME]
18    uv run portfolio.py remove TICKER [--portfolio NAME]
19"""
20 
21import argparse
22import json
23import os
24import sys
25from dataclasses import dataclass, asdict
26from datetime import datetime
27from pathlib import Path
28from typing import Literal
29 
30import yfinance as yf
31 
32 
33# Top 20 supported cryptocurrencies
34SUPPORTED_CRYPTOS = {
35    "BTC-USD", "ETH-USD", "BNB-USD", "SOL-USD", "XRP-USD",
36    "ADA-USD", "DOGE-USD", "AVAX-USD", "DOT-USD", "MATIC-USD",
37    "LINK-USD", "ATOM-USD", "UNI-USD", "LTC-USD", "BCH-USD",
38    "XLM-USD", "ALGO-USD", "VET-USD", "FIL-USD", "NEAR-USD",
39}
40 
41 
42def get_storage_path() -> Path:
43    """Get the portfolio storage path."""
44    # Use ~/.clawdbot/skills/stock-analysis/portfolios.json
45    state_dir = os.environ.get("CLAWDBOT_STATE_DIR", os.path.expanduser("~/.clawdbot"))
46    portfolio_dir = Path(state_dir) / "skills" / "stock-analysis"
47    portfolio_dir.mkdir(parents=True, exist_ok=True)
48    return portfolio_dir / "portfolios.json"
49 
50 
51def detect_asset_type(ticker: str) -> Literal["stock", "crypto"]:
52    """Detect asset type from ticker format."""
53    ticker_upper = ticker.upper()
54    if ticker_upper.endswith("-USD"):
55        base = ticker_upper[:-4]
56        if base.isalpha() and f"{base}-USD" in SUPPORTED_CRYPTOS:
57            return "crypto"
58        # Allow any *-USD ticker as crypto (flexible)
59        if base.isalpha():
60            return "crypto"
61    return "stock"
62 
63 
64@dataclass
65class Asset:
66    ticker: str
67    type: Literal["stock", "crypto"]
68    quantity: float
69    cost_basis: float
70    added_at: str
71 
72 
73@dataclass
74class Portfolio:
75    name: str
76    created_at: str
77    updated_at: str
78    assets: list[Asset]
79 
80 
81class PortfolioStore:
82    """Manages portfolio storage with atomic writes."""
83 
84    def __init__(self, path: Path | None = None):
85        self.path = path or get_storage_path()
86        self._data: dict | None = None
87 
88    def _load(self) -> dict:
89        """Load portfolios from disk."""
90        if self._data is not None:
91            return self._data
92 
93        if not self.path.exists():
94            self._data = {"version": 1, "portfolios": {}}
95            return self._data
96 
97        try:
98            with open(self.path, "r", encoding="utf-8") as f:
99                self._data = json.load(f)
100                return self._data
101        except (json.JSONDecodeError, IOError):
102            self._data = {"version": 1, "portfolios": {}}
103            return self._data
104 
105    def _save(self) -> None:
106        """Save portfolios to disk with atomic write."""
107        if self._data is None:
108            return
109 
110        # Ensure directory exists
111        self.path.parent.mkdir(parents=True, exist_ok=True)
112 
113        # Atomic write: write to temp file, then rename
114        tmp_path = self.path.with_suffix(".tmp")
115        try:
116            with open(tmp_path, "w", encoding="utf-8") as f:
117                json.dump(self._data, f, indent=2)
118            tmp_path.replace(self.path)
119        except Exception:
120            if tmp_path.exists():
121                tmp_path.unlink()
122            raise
123 
124    def _get_portfolio_key(self, name: str) -> str:
125        """Convert portfolio name to storage key."""
126        return name.lower().replace(" ", "-")
127 
128    def list_portfolios(self) -> list[str]:
129        """List all portfolio names."""
130        data = self._load()
131        return [p["name"] for p in data["portfolios"].values()]
132 
133    def get_portfolio(self, name: str) -> Portfolio | None:
134        """Get a portfolio by name."""
135        data = self._load()
136        key = self._get_portfolio_key(name)
137 
138        if key not in data["portfolios"]:
139            # Try case-insensitive match
140            for k, v in data["portfolios"].items():
141                if v["name"].lower() == name.lower():
142                    key = k
143                    break
144            else:
145                return None
146 
147        p = data["portfolios"][key]
148        assets = [
149            Asset(
150                ticker=a["ticker"],
151                type=a["type"],
152                quantity=a["quantity"],
153                cost_basis=a["cost_basis"],
154                added_at=a["added_at"],
155            )
156            for a in p.get("assets", [])
157        ]
158        return Portfolio(
159            name=p["name"],
160            created_at=p["created_at"],
161            updated_at=p["updated_at"],
162            assets=assets,
163        )
164 
165    def create_portfolio(self, name: str) -> Portfolio:
166        """Create a new portfolio."""
167        data = self._load()
168        key = self._get_portfolio_key(name)
169 
170        if key in data["portfolios"]:
171            raise ValueError(f"Portfolio '{name}' already exists")
172 
173        now = datetime.now().isoformat()
174        portfolio = {
175            "name": name,
176            "created_at": now,
177            "updated_at": now,
178            "assets": [],
179        }
180        data["portfolios"][key] = portfolio
181        self._save()
182 
183        return Portfolio(name=name, created_at=now, updated_at=now, assets=[])
184 
185    def delete_portfolio(self, name: str) -> bool:
186        """Delete a portfolio."""
187        data = self._load()
188        key = self._get_portfolio_key(name)
189 
190        # Try case-insensitive match
191        if key not in data["portfolios"]:
192            for k, v in data["portfolios"].items():
193                if v["name"].lower() == name.lower():
194                    key = k
195                    break
196            else:
197                return False
198 
199        del data["portfolios"][key]
200        self._save()
201        return True
202 
203    def rename_portfolio(self, old_name: str, new_name: str) -> bool:
204        """Rename a portfolio."""
205        data = self._load()
206        old_key = self._get_portfolio_key(old_name)
207        new_key = self._get_portfolio_key(new_name)
208 
209        # Find old portfolio
210        if old_key not in data["portfolios"]:
211            for k, v in data["portfolios"].items():
212                if v["name"].lower() == old_name.lower():
213                    old_key = k
214                    break
215            else:
216                return False
217 
218        if new_key in data["portfolios"] and new_key != old_key:
219            raise ValueError(f"Portfolio '{new_name}' already exists")
220 
221        portfolio = data["portfolios"].pop(old_key)
222        portfolio["name"] = new_name
223        portfolio["updated_at"] = datetime.now().isoformat()
224        data["portfolios"][new_key] = portfolio
225        self._save()
226        return True
227 
228    def add_asset(
229        self,
230        portfolio_name: str,
231        ticker: str,
232        quantity: float,
233        cost_basis: float,
234    ) -> Asset:
235        """Add an asset to a portfolio."""
236        data = self._load()
237        key = self._get_portfolio_key(portfolio_name)
238 
239        # Find portfolio
240        if key not in data["portfolios"]:
241            for k, v in data["portfolios"].items():
242                if v["name"].lower() == portfolio_name.lower():
243                    key = k
244                    break
245            else:
246                raise ValueError(f"Portfolio '{portfolio_name}' not found")
247 
248        portfolio = data["portfolios"][key]
249        ticker = ticker.upper()
250 
251        # Check if asset already exists
252        for asset in portfolio["assets"]:
253            if asset["ticker"] == ticker:
254                raise ValueError(f"Asset '{ticker}' already in portfolio. Use 'update' to modify.")
255 
256        # Validate ticker
257        asset_type = detect_asset_type(ticker)
258        try:
259            stock = yf.Ticker(ticker)
260            info = stock.info
261            if "regularMarketPrice" not in info:
262                raise ValueError(f"Invalid ticker: {ticker}")
263        except Exception as e:
264            raise ValueError(f"Could not validate ticker '{ticker}': {e}")
265 
266        now = datetime.now().isoformat()
267        asset = {
268            "ticker": ticker,
269            "type": asset_type,
270            "quantity": quantity,
271            "cost_basis": cost_basis,
272            "added_at": now,
273        }
274        portfolio["assets"].append(asset)
275        portfolio["updated_at"] = now
276        self._save()
277 
278        return Asset(**asset)
279 
280    def update_asset(
281        self,
282        portfolio_name: str,
283        ticker: str,
284        quantity: float | None = None,
285        cost_basis: float | None = None,
286    ) -> Asset | None:
287        """Update an asset in a portfolio."""
288        data = self._load()
289        key = self._get_portfolio_key(portfolio_name)
290 
291        # Find portfolio
292        if key not in data["portfolios"]:
293            for k, v in data["portfolios"].items():
294                if v["name"].lower() == portfolio_name.lower():
295                    key = k
296                    break
297            else:
298                return None
299 
300        portfolio = data["portfolios"][key]
301        ticker = ticker.upper()
302 
303        for asset in portfolio["assets"]:
304            if asset["ticker"] == ticker:
305                if quantity is not None:
306                    asset["quantity"] = quantity
307                if cost_basis is not None:
308                    asset["cost_basis"] = cost_basis
309                portfolio["updated_at"] = datetime.now().isoformat()
310                self._save()
311                return Asset(**asset)
312 
313        return None
314 
315    def remove_asset(self, portfolio_name: str, ticker: str) -> bool:
316        """Remove an asset from a portfolio."""
317        data = self._load()
318        key = self._get_portfolio_key(portfolio_name)
319 
320        # Find portfolio
321        if key not in data["portfolios"]:
322            for k, v in data["portfolios"].items():
323                if v["name"].lower() == portfolio_name.lower():
324                    key = k
325                    break
326            else:
327                return False
328 
329        portfolio = data["portfolios"][key]
330        ticker = ticker.upper()
331 
332        original_len = len(portfolio["assets"])
333        portfolio["assets"] = [a for a in portfolio["assets"] if a["ticker"] != ticker]
334 
335        if len(portfolio["assets"]) < original_len:
336            portfolio["updated_at"] = datetime.now().isoformat()
337            self._save()
338            return True
339 
340        return False
341 
342    def get_default_portfolio_name(self) -> str | None:
343        """Get the default (first) portfolio name, or None if empty."""
344        portfolios = self.list_portfolios()
345        return portfolios[0] if portfolios else None
346 
347 
348def format_currency(value: float) -> str:
349    """Format a value as currency."""
350    if abs(value) >= 1_000_000:
351        return f"${value/1_000_000:.2f}M"
352    elif abs(value) >= 1_000:
353        return f"${value/1_000:.2f}K"
354    else:
355        return f"${value:.2f}"
356 
357 
358def show_portfolio(portfolio: Portfolio, verbose: bool = False) -> None:
359    """Display portfolio details with current prices."""
360    print(f"\n{'='*60}")
361    print(f"PORTFOLIO: {portfolio.name}")
362    print(f"Created: {portfolio.created_at[:10]} | Updated: {portfolio.updated_at[:10]}")
363    print(f"{'='*60}\n")
364 
365    if not portfolio.assets:
366        print("  No assets in portfolio. Use 'add' to add assets.\n")
367        return
368 
369    total_cost = 0.0
370    total_value = 0.0
371 
372    print(f"{'Ticker':<12} {'Type':<8} {'Qty':>10} {'Cost':>12} {'Current':>12} {'Value':>14} {'P&L':>12}")
373    print("-" * 82)
374 
375    for asset in portfolio.assets:
376        try:
377            stock = yf.Ticker(asset.ticker)
378            current_price = stock.info.get("regularMarketPrice", 0) or 0
379        except Exception:
380            current_price = 0
381 
382        cost_total = asset.quantity * asset.cost_basis
383        current_value = asset.quantity * current_price
384        pnl = current_value - cost_total
385        pnl_pct = (pnl / cost_total * 100) if cost_total > 0 else 0
386 
387        total_cost += cost_total
388        total_value += current_value
389 
390        pnl_str = f"{'+' if pnl >= 0 else ''}{format_currency(pnl)} ({pnl_pct:+.1f}%)"
391 
392        print(f"{asset.ticker:<12} {asset.type:<8} {asset.quantity:>10.4f} "
393              f"{format_currency(asset.cost_basis):>12} {format_currency(current_price):>12} "
394              f"{format_currency(current_value):>14} {pnl_str:>12}")
395 
396    print("-" * 82)
397    total_pnl = total_value - total_cost
398    total_pnl_pct = (total_pnl / total_cost * 100) if total_cost > 0 else 0
399    print(f"{'TOTAL':<12} {'':<8} {'':<10} {format_currency(total_cost):>12} {'':<12} "
400          f"{format_currency(total_value):>14} {'+' if total_pnl >= 0 else ''}{format_currency(total_pnl)} ({total_pnl_pct:+.1f}%)")
401    print()
402 
403 
404def main():
405    parser = argparse.ArgumentParser(description="Portfolio management for stock-analysis")
406    subparsers = parser.add_subparsers(dest="command", help="Commands")
407 
408    # create
409    create_parser = subparsers.add_parser("create", help="Create a new portfolio")
410    create_parser.add_argument("name", help="Portfolio name")
411 
412    # list
413    subparsers.add_parser("list", help="List all portfolios")
414 
415    # show
416    show_parser = subparsers.add_parser("show", help="Show portfolio details")
417    show_parser.add_argument("--portfolio", "-p", help="Portfolio name (default: first portfolio)")
418 
419    # delete
420    delete_parser = subparsers.add_parser("delete", help="Delete a portfolio")
421    delete_parser.add_argument("name", help="Portfolio name")
422 
423    # rename
424    rename_parser = subparsers.add_parser("rename", help="Rename a portfolio")
425    rename_parser.add_argument("old_name", help="Current portfolio name")
426    rename_parser.add_argument("new_name", help="New portfolio name")
427 
428    # add
429    add_parser = subparsers.add_parser("add", help="Add an asset to portfolio")
430    add_parser.add_argument("ticker", help="Stock/crypto ticker (e.g., AAPL, BTC-USD)")
431    add_parser.add_argument("--quantity", "-q", type=float, required=True, help="Quantity")
432    add_parser.add_argument("--cost", "-c", type=float, required=True, help="Cost basis per unit")
433    add_parser.add_argument("--portfolio", "-p", help="Portfolio name (default: first portfolio)")
434 
435    # update
436    update_parser = subparsers.add_parser("update", help="Update an asset in portfolio")
437    update_parser.add_argument("ticker", help="Stock/crypto ticker")
438    update_parser.add_argument("--quantity", "-q", type=float, help="New quantity")
439    update_parser.add_argument("--cost", "-c", type=float, help="New cost basis per unit")
440    update_parser.add_argument("--portfolio", "-p", help="Portfolio name (default: first portfolio)")
441 
442    # remove
443    remove_parser = subparsers.add_parser("remove", help="Remove an asset from portfolio")
444    remove_parser.add_argument("ticker", help="Stock/crypto ticker")
445    remove_parser.add_argument("--portfolio", "-p", help="Portfolio name (default: first portfolio)")
446 
447    args = parser.parse_args()
448 
449    if not args.command:
450        parser.print_help()
451        sys.exit(1)
452 
453    store = PortfolioStore()
454 
455    try:
456        if args.command == "create":
457            portfolio = store.create_portfolio(args.name)
458            print(f"Created portfolio: {portfolio.name}")
459 
460        elif args.command == "list":
461            portfolios = store.list_portfolios()
462            if not portfolios:
463                print("No portfolios found. Use 'create' to create one.")
464            else:
465                print("\nPortfolios:")
466                for name in portfolios:
467                    p = store.get_portfolio(name)
468                    asset_count = len(p.assets) if p else 0
469                    print(f"  - {name} ({asset_count} assets)")
470                print()
471 
472        elif args.command == "show":
473            portfolio_name = args.portfolio or store.get_default_portfolio_name()
474            if not portfolio_name:
475                print("No portfolios found. Use 'create' to create one.")
476                sys.exit(1)
477 
478            portfolio = store.get_portfolio(portfolio_name)
479            if not portfolio:
480                print(f"Portfolio '{portfolio_name}' not found.")
481                sys.exit(1)
482 
483            show_portfolio(portfolio)
484 
485        elif args.command == "delete":
486            if store.delete_portfolio(args.name):
487                print(f"Deleted portfolio: {args.name}")
488            else:
489                print(f"Portfolio '{args.name}' not found.")
490                sys.exit(1)
491 
492        elif args.command == "rename":
493            if store.rename_portfolio(args.old_name, args.new_name):
494                print(f"Renamed portfolio: {args.old_name} -> {args.new_name}")
495            else:
496                print(f"Portfolio '{args.old_name}' not found.")
497                sys.exit(1)
498 
499        elif args.command == "add":
500            portfolio_name = args.portfolio or store.get_default_portfolio_name()
501            if not portfolio_name:
502                print("No portfolios found. Use 'create' to create one first.")
503                sys.exit(1)
504 
505            asset = store.add_asset(portfolio_name, args.ticker, args.quantity, args.cost)
506            print(f"Added {asset.ticker} ({asset.type}) to {portfolio_name}: "
507                  f"{asset.quantity} units @ {format_currency(asset.cost_basis)}")
508 
509        elif args.command == "update":
510            portfolio_name = args.portfolio or store.get_default_portfolio_name()
511            if not portfolio_name:
512                print("No portfolios found.")
513                sys.exit(1)
514 
515            if args.quantity is None and args.cost is None:
516                print("Must specify --quantity and/or --cost to update.")
517                sys.exit(1)
518 
519            asset = store.update_asset(portfolio_name, args.ticker, args.quantity, args.cost)
520            if asset:
521                print(f"Updated {asset.ticker} in {portfolio_name}: "
522                      f"{asset.quantity} units @ {format_currency(asset.cost_basis)}")
523            else:
524                print(f"Asset '{args.ticker}' not found in portfolio '{portfolio_name}'.")
525                sys.exit(1)
526 
527        elif args.command == "remove":
528            portfolio_name = args.portfolio or store.get_default_portfolio_name()
529            if not portfolio_name:
530                print("No portfolios found.")
531                sys.exit(1)
532 
533            if store.remove_asset(portfolio_name, args.ticker):
534                print(f"Removed {args.ticker.upper()} from {portfolio_name}")
535            else:
536                print(f"Asset '{args.ticker}' not found in portfolio '{portfolio_name}'.")
537                sys.exit(1)
538 
539    except ValueError as e:
540        print(f"Error: {e}")
541        sys.exit(1)
542    except Exception as e:
543        print(f"Unexpected error: {e}")
544        sys.exit(1)
545 
546 
547if __name__ == "__main__":
548    main()
549