Tracking Crypto Portfolio

# Crypto Portfolio Tracker Configuration
# Version: 2.0.0

# =============================================================================
# Display Settings
# =============================================================================

display:
  default_format: table          # table, json, csv
  table_width: 78
  show_all_holdings: false       # Show top 5 by default
  default_sort: value            # value, allocation, name, change
  currency: USD

# =============================================================================
# Allocation Thresholds
# =============================================================================

thresholds:
  # Concentration warning threshold (percent)
  concentration_warning: 25.0

  # Risk levels for reporting
  risk_levels:
    low: 10                      # < 10% is low risk
    medium: 25                   # 10-25% is medium
    high: 50                     # 25-50% is high
    # > 50% is very high

# =============================================================================
# API Configuration
# =============================================================================

api:
  coingecko:
    base_url: "https://api.coingecko.com/api/v3"
    timeout: 15                  # Request timeout in seconds
    rate_limit: 30               # Max requests per minute (free tier)
    batch_size: 250              # Max coins per request

# =============================================================================
# Caching
# =============================================================================

cache:
  enabled: true
  ttl: 60                        # Time-to-live in seconds
  file: ".price_cache.json"      # Cache file (relative to scripts/)

# =============================================================================
# Coin Symbol Mappings
# =============================================================================
# Maps common symbols to CoinGecko IDs
# Add custom mappings here for non-standard symbols

symbol_mappings:
  # Layer 1
  BTC: bitcoin
  ETH: ethereum
  SOL: solana
  ADA: cardano
  DOT: polkadot
  AVAX: avalanche-2
  NEAR: near
  APT: aptos
  SUI: sui

  # Layer 2 / Scaling
  MATIC: matic-network
  ARB: arbitrum
  OP: optimism

  # DeFi
  LINK: chainlink
  UNI: uniswap
  AAVE: aave
  MKR: maker
  CRV: curve-dao-token
  LDO: lido-dao

  # Stablecoins
  USDC: usd-coin
  USDT: tether
  DAI: dai
  FRAX: frax

  # Meme / Community
  DOGE: dogecoin
  SHIB: shiba-inu
  PEPE: pepe
  BONK: bonk

  # Other Major
  XRP: ripple
  LTC: litecoin
  BNB: binancecoin
  ATOM: cosmos

# =============================================================================
# Category Definitions
# =============================================================================
# Default category mappings if not specified in portfolio file

default_categories:
  # Layer 1
  BTC: "Layer 1"
  ETH: "Layer 1"
  SOL: "Layer 1"
  ADA: "Layer 1"
  DOT: "Layer 1"
  AVAX: "Layer 1"

  # DeFi
  LINK: "DeFi"
  UNI: "DeFi"
  AAVE: "DeFi"
  MKR: "DeFi"
  CRV: "DeFi"

  # Stablecoins
  USDC: "Stablecoin"
  USDT: "Stablecoin"
  DAI: "Stablecoin"

  # Meme
  DOGE: "Meme"
  SHIB: "Meme"
  PEPE: "Meme"

# =============================================================================
# Portfolio File Schema
# =============================================================================
# Reference for portfolio file structure

portfolio_schema:
  required_fields:
    - coin                       # Symbol (BTC, ETH, etc.)
    - quantity                   # Amount held

  optional_fields:
    - cost_basis                 # Average purchase price per coin
    - acquired                   # Date acquired (YYYY-MM-DD)
    - wallet                     # Wallet/exchange name
    - notes                      # Free-form notes

  portfolio_fields:
    - name                       # Portfolio name
    - currency                   # Display currency (default: USD)
    - categories                 # Custom category mappings

# =============================================================================
# Output Formatting
# =============================================================================

formatting:
  decimal_places:
    price: 2
    quantity: 4
    value: 2
    percent: 2
    allocation: 1

  # ANSI colors for terminal output
  colors:
    positive: "\033[92m"         # Green
    negative: "\033[91m"         # Red
    warning: "\033[93m"          # Yellow
    reset: "\033[0m"

#!/usr/bin/env python3
"""
Portfolio Output Formatters

Formats portfolio data in various output formats:
- Table (default): Terminal-friendly dashboard
- JSON: Machine-readable export
- CSV: Spreadsheet-compatible

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import csv
import io
import json
from datetime import datetime
from typing import Dict, Any, List


class PortfolioFormatter:
    """Formats portfolio data for output."""

    def format(
        self,
        data: Dict[str, Any],
        format_type: str = "table",
        show_all_holdings: bool = False,
        show_pnl: bool = False
    ) -> str:
        """Format portfolio data for output.

        Args:
            data: Portfolio valuation data
            format_type: Output format ("table", "json", "csv")
            show_all_holdings: Show all holdings (not just top 5)
            show_pnl: Show P&L information

        Returns:
            Formatted string output
        """
        if format_type == "json":
            return self._format_json(data)
        elif format_type == "csv":
            return self._format_csv(data, show_pnl)
        else:
            return self._format_table(data, show_all_holdings, show_pnl)

    def _format_table(
        self,
        data: Dict[str, Any],
        show_all: bool,
        show_pnl: bool
    ) -> str:
        """Format as terminal table/dashboard."""
        lines = []
        w = 78  # Width

        # Header
        timestamp = data.get("meta", {}).get("timestamp", "")
        if timestamp:
            try:
                dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
                time_str = dt.strftime("%Y-%m-%d %H:%M UTC")
            except ValueError:
                time_str = timestamp[:19]
        else:
            time_str = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")

        lines.append("=" * w)
        lines.append(f"  CRYPTO PORTFOLIO TRACKER{' ' * (w - 41)}{time_str}")
        lines.append("=" * w)
        lines.append("")

        # Portfolio Summary
        name = data.get("portfolio_name", "My Portfolio")
        total = data.get("total_value_usd", 0)
        holdings_count = data.get("holdings_count", 0)

        lines.append(f"  PORTFOLIO SUMMARY: {name}")
        lines.append("-" * w)
        lines.append(f"  Total Value:    ${total:,.2f} USD")

        # 24h change
        change_24h = data.get("change_24h", {})
        if change_24h.get("percent") is not None:
            amount = change_24h.get("amount", 0)
            pct = change_24h.get("percent", 0)
            sign = "+" if amount >= 0 else ""
            lines.append(f"  24h Change:     {sign}${amount:,.2f} ({sign}{pct:.2f}%)")

        # 7d change
        change_7d = data.get("change_7d", {})
        if change_7d.get("percent") is not None:
            amount = change_7d.get("amount", 0)
            pct = change_7d.get("percent", 0)
            sign = "+" if amount >= 0 else ""
            lines.append(f"  7d Change:      {sign}${amount:,.2f} ({sign}{pct:.2f}%)")

        lines.append(f"  Holdings:       {holdings_count} assets")

        # Total P&L if available
        if show_pnl and data.get("total_unrealized_pnl") is not None:
            pnl = data.get("total_unrealized_pnl", 0)
            pnl_pct = data.get("total_pnl_pct", 0)
            sign = "+" if pnl >= 0 else ""
            lines.append(f"  Total P&L:      {sign}${pnl:,.2f} ({sign}{pnl_pct:.2f}%)")

        lines.append("")

        # Holdings Table
        holdings = data.get("holdings", [])
        if holdings:
            display_count = len(holdings) if show_all else min(5, len(holdings))

            if show_all:
                lines.append("  ALL HOLDINGS")
            else:
                lines.append("  TOP HOLDINGS")
            lines.append("-" * w)

            # Table header
            if show_pnl:
                lines.append(f"  {'Coin':<6} {'Quantity':>12} {'Price':>12} {'Value':>14} {'Alloc':>7} {'P&L':>10}")
            else:
                lines.append(f"  {'Coin':<6} {'Quantity':>12} {'Price':>12} {'Value':>14} {'Alloc':>7} {'24h':>8}")

            for holding in holdings[:display_count]:
                coin = holding.get("coin", "")
                qty = holding.get("quantity", 0)
                price = holding.get("price_usd", 0)
                value = holding.get("value_usd", 0)
                alloc = holding.get("allocation_pct", 0)

                if show_pnl and holding.get("pnl_pct") is not None:
                    pnl_pct = holding.get("pnl_pct", 0)
                    sign = "+" if pnl_pct >= 0 else ""
                    change_str = f"{sign}{pnl_pct:.1f}%"
                else:
                    change_24h = holding.get("change_24h_pct")
                    if change_24h is not None:
                        sign = "+" if change_24h >= 0 else ""
                        change_str = f"{sign}{change_24h:.1f}%"
                    else:
                        change_str = "N/A"

                lines.append(
                    f"  {coin:<6} {qty:>12.4f} ${price:>10,.2f} ${value:>12,.2f} {alloc:>6.1f}% {change_str:>8}"
                )

            if not show_all and len(holdings) > 5:
                remaining = len(holdings) - 5
                lines.append(f"  ... and {remaining} more holdings")

            lines.append("")

        # Category Allocation
        categories = data.get("allocation_by_category", {})
        if categories and show_all:
            lines.append("  ALLOCATION BY CATEGORY")
            lines.append("-" * w)
            for category, pct in categories.items():
                bar_len = int(pct / 2)  # Scale to fit
                bar = "█" * bar_len
                lines.append(f"  {category:<15} {pct:>5.1f}% {bar}")
            lines.append("")

        # Risk Flags
        risk_flags = data.get("risk_flags", [])
        if risk_flags:
            lines.append("  ⚠ CONCENTRATION WARNINGS")
            lines.append("-" * w)
            for flag in risk_flags:
                lines.append(f"  • {flag}")
            lines.append("")

        # Footer
        lines.append("=" * w)
        threshold = data.get("meta", {}).get("threshold", 25)
        lines.append(f"  Concentration threshold: {threshold}%  |  Data: CoinGecko")
        lines.append("=" * w)

        return "\n".join(lines)

    def _format_json(self, data: Dict[str, Any]) -> str:
        """Format as JSON."""
        return json.dumps(data, indent=2, default=str)

    def _format_csv(self, data: Dict[str, Any], include_pnl: bool) -> str:
        """Format as CSV."""
        output = io.StringIO()
        writer = csv.writer(output)

        # Header row
        headers = [
            "coin",
            "quantity",
            "price_usd",
            "value_usd",
            "allocation_pct",
            "change_24h_pct",
            "change_7d_pct",
            "market_cap"
        ]

        if include_pnl:
            headers.extend([
                "cost_basis",
                "total_cost",
                "unrealized_pnl",
                "pnl_pct"
            ])

        writer.writerow(headers)

        # Data rows
        for holding in data.get("holdings", []):
            row = [
                holding.get("coin", ""),
                holding.get("quantity", 0),
                holding.get("price_usd", 0),
                holding.get("value_usd", 0),
                holding.get("allocation_pct", 0),
                holding.get("change_24h_pct", ""),
                holding.get("change_7d_pct", ""),
                holding.get("market_cap", "")
            ]

            if include_pnl:
                row.extend([
                    holding.get("cost_basis", ""),
                    holding.get("total_cost", ""),
                    holding.get("unrealized_pnl", ""),
                    holding.get("pnl_pct", "")
                ])

            writer.writerow(row)

        # Summary row
        writer.writerow([])
        writer.writerow(["SUMMARY"])
        writer.writerow(["Portfolio Name", data.get("portfolio_name", "")])
        writer.writerow(["Total Value USD", data.get("total_value_usd", 0)])
        writer.writerow(["Holdings Count", data.get("holdings_count", 0)])

        if data.get("change_24h", {}).get("percent") is not None:
            writer.writerow(["24h Change %", data["change_24h"]["percent"]])

        if include_pnl and data.get("total_unrealized_pnl") is not None:
            writer.writerow(["Total Unrealized P&L", data.get("total_unrealized_pnl", 0)])
            writer.writerow(["Total P&L %", data.get("total_pnl_pct", 0)])

        return output.getvalue()


def main():
    """CLI entry point for testing."""
    # Test with sample data
    sample_data = {
        "portfolio_name": "Test Portfolio",
        "total_value_usd": 125450.00,
        "change_24h": {"amount": 2540.50, "percent": 2.07},
        "change_7d": {"amount": 8125.00, "percent": 6.92},
        "holdings_count": 5,
        "total_cost": 80000,
        "total_unrealized_pnl": 45450,
        "total_pnl_pct": 56.8,
        "holdings": [
            {
                "coin": "BTC",
                "quantity": 0.5,
                "price_usd": 95000,
                "value_usd": 47500,
                "allocation_pct": 37.9,
                "change_24h_pct": 2.5,
                "cost_basis": 50000,
                "total_cost": 25000,
                "unrealized_pnl": 22500,
                "pnl_pct": 90.0
            },
            {
                "coin": "ETH",
                "quantity": 10,
                "price_usd": 3200,
                "value_usd": 32000,
                "allocation_pct": 25.5,
                "change_24h_pct": 1.8,
                "cost_basis": 2500,
                "total_cost": 25000,
                "unrealized_pnl": 7000,
                "pnl_pct": 28.0
            },
            {
                "coin": "SOL",
                "quantity": 100,
                "price_usd": 180,
                "value_usd": 18000,
                "allocation_pct": 14.4,
                "change_24h_pct": 4.2
            }
        ],
        "allocation_by_category": {
            "Layer 1": 77.8,
            "Other": 22.2
        },
        "risk_flags": [
            "BTC allocation (37.9%) exceeds 25% threshold",
            "ETH allocation (25.5%) exceeds 25% threshold"
        ],
        "meta": {
            "timestamp": "2026-01-14T15:30:00Z",
            "threshold": 25
        }
    }

    formatter = PortfolioFormatter()

    print("=== TABLE FORMAT ===")
    print(formatter.format(sample_data, "table", show_all_holdings=True, show_pnl=True))
    print()
    print("=== JSON FORMAT ===")
    print(formatter.format(sample_data, "json"))
    print()
    print("=== CSV FORMAT ===")
    print(formatter.format(sample_data, "csv", show_pnl=True))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Portfolio Loader

Loads and validates portfolio JSON files.
Supports multiple portfolio formats with graceful handling.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import json
import sys
from pathlib import Path
from typing import Optional, Dict, Any, List


class PortfolioLoader:
    """Loads and validates portfolio files."""

    def __init__(self, verbose: bool = False):
        """Initialize loader.

        Args:
            verbose: Enable verbose output
        """
        self.verbose = verbose

    def load(self, path: str) -> Optional[Dict[str, Any]]:
        """Load portfolio from JSON file.

        Args:
            path: Path to portfolio JSON file

        Returns:
            Validated portfolio dict or None on error
        """
        file_path = Path(path).expanduser()

        # Check file exists
        if not file_path.exists():
            print(f"Error: Portfolio file not found: {file_path}", file=sys.stderr)
            return None

        # Load JSON
        try:
            with open(file_path, "r") as f:
                data = json.load(f)
        except json.JSONDecodeError as e:
            print(f"Error: Invalid JSON in portfolio file: {e}", file=sys.stderr)
            return None
        except Exception as e:
            print(f"Error: Failed to read portfolio file: {e}", file=sys.stderr)
            return None

        # Validate and normalize
        return self._validate_portfolio(data)

    def _validate_portfolio(self, data: Any) -> Optional[Dict[str, Any]]:
        """Validate and normalize portfolio data."""
        # Handle list format (just holdings array)
        if isinstance(data, list):
            data = {"holdings": data}

        if not isinstance(data, dict):
            print("Error: Portfolio must be a JSON object or array", file=sys.stderr)
            return None

        # Check for holdings
        holdings = data.get("holdings", [])
        if not holdings:
            print("Error: Portfolio has no holdings", file=sys.stderr)
            return None

        # Validate each holding
        valid_holdings = []
        for i, holding in enumerate(holdings):
            validated = self._validate_holding(holding, i)
            if validated:
                valid_holdings.append(validated)

        if not valid_holdings:
            print("Error: No valid holdings found in portfolio", file=sys.stderr)
            return None

        # Aggregate duplicate coins
        aggregated = self._aggregate_holdings(valid_holdings)

        return {
            "name": data.get("name", "My Portfolio"),
            "holdings": aggregated,
            "categories": data.get("categories", {}),
            "currency": data.get("currency", "USD")
        }

    def _validate_holding(self, holding: Any, index: int) -> Optional[Dict[str, Any]]:
        """Validate a single holding entry."""
        if not isinstance(holding, dict):
            if self.verbose:
                print(f"Warning: Holding {index} is not an object, skipping", file=sys.stderr)
            return None

        # Required: coin symbol
        coin = holding.get("coin") or holding.get("symbol") or holding.get("ticker")
        if not coin:
            if self.verbose:
                print(f"Warning: Holding {index} missing coin symbol, skipping", file=sys.stderr)
            return None

        coin = str(coin).upper().strip()

        # Required: quantity
        quantity = holding.get("quantity") or holding.get("amount") or holding.get("qty")
        if quantity is None:
            if self.verbose:
                print(f"Warning: Holding {index} ({coin}) missing quantity, skipping", file=sys.stderr)
            return None

        try:
            quantity = float(quantity)
        except (TypeError, ValueError):
            if self.verbose:
                print(f"Warning: Holding {index} ({coin}) has invalid quantity, skipping", file=sys.stderr)
            return None

        if quantity <= 0:
            if self.verbose:
                print(f"Warning: Holding {index} ({coin}) has non-positive quantity, skipping", file=sys.stderr)
            return None

        # Optional: cost basis
        cost_basis = holding.get("cost_basis") or holding.get("cost") or holding.get("avg_cost")
        if cost_basis is not None:
            try:
                cost_basis = float(cost_basis)
                if cost_basis < 0:
                    cost_basis = None
            except (TypeError, ValueError):
                cost_basis = None

        # Optional: acquired date
        acquired = holding.get("acquired") or holding.get("date") or holding.get("purchase_date")

        # Optional: wallet/location
        wallet = holding.get("wallet") or holding.get("location") or holding.get("exchange")

        # Optional: notes
        notes = holding.get("notes") or holding.get("memo")

        return {
            "coin": coin,
            "quantity": quantity,
            "cost_basis": cost_basis,
            "acquired": acquired,
            "wallet": wallet,
            "notes": notes
        }

    def _aggregate_holdings(self, holdings: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Aggregate holdings for the same coin."""
        aggregated = {}

        for holding in holdings:
            coin = holding["coin"]

            if coin not in aggregated:
                aggregated[coin] = {
                    "coin": coin,
                    "quantity": 0,
                    "total_cost": 0,
                    "has_cost_basis": False,
                    "wallets": [],
                    "notes": []
                }

            aggregated[coin]["quantity"] += holding["quantity"]

            # Track cost basis if available
            if holding.get("cost_basis") is not None:
                cost = holding["quantity"] * holding["cost_basis"]
                aggregated[coin]["total_cost"] += cost
                aggregated[coin]["has_cost_basis"] = True

            # Track wallets
            if holding.get("wallet"):
                aggregated[coin]["wallets"].append(holding["wallet"])

            # Track notes
            if holding.get("notes"):
                aggregated[coin]["notes"].append(holding["notes"])

        # Convert to list with calculated average cost basis
        result = []
        for coin, data in aggregated.items():
            entry = {
                "coin": coin,
                "quantity": data["quantity"]
            }

            # Calculate average cost basis
            if data["has_cost_basis"] and data["quantity"] > 0:
                entry["cost_basis"] = data["total_cost"] / data["quantity"]

            # Include wallets if tracked
            if data["wallets"]:
                entry["wallets"] = list(set(data["wallets"]))

            # Include notes
            if data["notes"]:
                entry["notes"] = "; ".join(data["notes"])

            result.append(entry)

        return result


def main():
    """CLI entry point for testing."""
    import argparse

    parser = argparse.ArgumentParser(description="Load and validate portfolio file")
    parser.add_argument("portfolio", type=str, help="Path to portfolio JSON")
    parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
    args = parser.parse_args()

    loader = PortfolioLoader(verbose=args.verbose)
    portfolio = loader.load(args.portfolio)

    if portfolio:
        print(json.dumps(portfolio, indent=2))
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Crypto Portfolio Tracker - Main CLI Entry Point

Track cryptocurrency portfolio with real-time valuations,
allocation analysis, and P&L tracking.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List

# Add scripts directory to path for local imports
SCRIPT_DIR = Path(__file__).parent
sys.path.insert(0, str(SCRIPT_DIR))

from portfolio_loader import PortfolioLoader
from price_fetcher import PriceFetcher
from valuation_engine import ValuationEngine
from formatters import PortfolioFormatter


def parse_args() -> argparse.Namespace:
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Track cryptocurrency portfolio with real-time valuations",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --portfolio holdings.json              # Portfolio summary
  %(prog)s --portfolio holdings.json --holdings   # All holdings
  %(prog)s --portfolio holdings.json --detailed   # Full analysis
  %(prog)s --portfolio holdings.json --format json  # JSON export
        """
    )

    # Required
    parser.add_argument(
        "--portfolio", "-p",
        type=str,
        required=True,
        help="Path to portfolio JSON file"
    )

    # Display options
    parser.add_argument(
        "--holdings",
        action="store_true",
        help="Show all holdings breakdown"
    )
    parser.add_argument(
        "--detailed",
        action="store_true",
        help="Show detailed analysis with P&L"
    )
    parser.add_argument(
        "--sort",
        type=str,
        choices=["value", "allocation", "name", "change"],
        default="value",
        help="Sort holdings by (default: value)"
    )

    # Thresholds
    parser.add_argument(
        "--threshold",
        type=float,
        default=25.0,
        help="Allocation warning threshold in percent (default: 25)"
    )

    # Output options
    parser.add_argument(
        "--format", "-f",
        type=str,
        choices=["table", "json", "csv"],
        default="table",
        help="Output format (default: table)"
    )
    parser.add_argument(
        "--output", "-o",
        type=str,
        help="Output file path (default: stdout)"
    )

    # Debug options
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Enable verbose output"
    )
    parser.add_argument(
        "--version",
        action="version",
        version="%(prog)s 2.0.0"
    )

    return parser.parse_args()


def main() -> None:
    """Main entry point."""
    args = parse_args()

    # Initialize components
    loader = PortfolioLoader(verbose=args.verbose)
    fetcher = PriceFetcher(verbose=args.verbose)
    engine = ValuationEngine(verbose=args.verbose)
    formatter = PortfolioFormatter()

    # Load portfolio
    if args.verbose:
        print(f"Loading portfolio from {args.portfolio}...", file=sys.stderr)

    portfolio = loader.load(args.portfolio)
    if not portfolio:
        print(f"Error: Failed to load portfolio from {args.portfolio}", file=sys.stderr)
        sys.exit(1)

    if args.verbose:
        print(f"Loaded {len(portfolio.get('holdings', []))} holdings", file=sys.stderr)

    # Get unique coins
    coins = list(set(h.get("coin", "").upper() for h in portfolio.get("holdings", [])))
    coins = [c for c in coins if c]  # Remove empty

    if not coins:
        print("Error: No valid holdings found in portfolio", file=sys.stderr)
        sys.exit(1)

    # Fetch prices
    if args.verbose:
        print(f"Fetching prices for {len(coins)} coins...", file=sys.stderr)

    prices = fetcher.fetch_prices(coins)
    if not prices:
        print("Warning: Could not fetch prices, using cached or fallback values", file=sys.stderr)

    # Calculate valuations
    if args.verbose:
        print("Calculating valuations...", file=sys.stderr)

    valuations = engine.calculate(
        portfolio=portfolio,
        prices=prices,
        threshold=args.threshold
    )

    # Sort holdings
    if valuations.get("holdings"):
        sort_key = {
            "value": lambda x: x.get("value_usd", 0),
            "allocation": lambda x: x.get("allocation_pct", 0),
            "name": lambda x: x.get("coin", ""),
            "change": lambda x: x.get("change_24h_pct", 0) or 0,
        }.get(args.sort, lambda x: x.get("value_usd", 0))

        reverse = args.sort != "name"
        valuations["holdings"] = sorted(
            valuations["holdings"],
            key=sort_key,
            reverse=reverse
        )

    # Add metadata
    valuations["meta"] = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "portfolio_file": args.portfolio,
        "threshold": args.threshold,
        "sort_by": args.sort
    }

    # Determine display mode
    show_all = args.holdings or args.detailed
    show_pnl = args.detailed

    # Format output
    output = formatter.format(
        valuations,
        format_type=args.format,
        show_all_holdings=show_all,
        show_pnl=show_pnl
    )

    # Write output
    if args.output:
        output_path = Path(args.output)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, "w") as f:
            f.write(output)
        print(f"Output written to {output_path}", file=sys.stderr)
    else:
        print(output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Price Fetcher

Fetches cryptocurrency prices from CoinGecko API.
Implements caching and batch requests for efficiency.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import json
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List

try:
    import requests
except ImportError:
    print("Error: requests library required. Install with: pip install requests", file=sys.stderr)
    sys.exit(1)


# Map common symbols to CoinGecko IDs
SYMBOL_TO_ID = {
    "BTC": "bitcoin",
    "ETH": "ethereum",
    "SOL": "solana",
    "XRP": "ripple",
    "ADA": "cardano",
    "DOGE": "dogecoin",
    "DOT": "polkadot",
    "LINK": "chainlink",
    "AVAX": "avalanche-2",
    "MATIC": "matic-network",
    "BNB": "binancecoin",
    "LTC": "litecoin",
    "ATOM": "cosmos",
    "UNI": "uniswap",
    "AAVE": "aave",
    "CRV": "curve-dao-token",
    "MKR": "maker",
    "COMP": "compound-governance-token",
    "SNX": "havven",
    "SUSHI": "sushi",
    "YFI": "yearn-finance",
    "1INCH": "1inch",
    "ENS": "ethereum-name-service",
    "LDO": "lido-dao",
    "ARB": "arbitrum",
    "OP": "optimism",
    "APT": "aptos",
    "SUI": "sui",
    "SEI": "sei-network",
    "TIA": "celestia",
    "INJ": "injective-protocol",
    "FET": "fetch-ai",
    "RNDR": "render-token",
    "GRT": "the-graph",
    "FIL": "filecoin",
    "NEAR": "near",
    "ICP": "internet-computer",
    "HBAR": "hedera-hashgraph",
    "VET": "vechain",
    "ALGO": "algorand",
    "XLM": "stellar",
    "XTZ": "tezos",
    "EOS": "eos",
    "FLOW": "flow",
    "MANA": "decentraland",
    "SAND": "the-sandbox",
    "AXS": "axie-infinity",
    "APE": "apecoin",
    "SHIB": "shiba-inu",
    "PEPE": "pepe",
    "WIF": "dogwifcoin",
    "BONK": "bonk",
    # Stablecoins
    "USDC": "usd-coin",
    "USDT": "tether",
    "DAI": "dai",
    "BUSD": "binance-usd",
    "TUSD": "true-usd",
    "FRAX": "frax",
    "LUSD": "liquity-usd",
    "USDD": "usdd",
}


class PriceFetcher:
    """Fetches cryptocurrency prices from CoinGecko."""

    COINGECKO_API = "https://api.coingecko.com/api/v3"
    CACHE_FILE = Path(__file__).parent / ".price_cache.json"
    CACHE_TTL = 60  # 1 minute

    def __init__(self, verbose: bool = False):
        """Initialize fetcher.

        Args:
            verbose: Enable verbose output
        """
        self.verbose = verbose
        self._cache: Dict[str, Any] = {}
        self._cache_time: float = 0

    def fetch_prices(self, symbols: List[str]) -> Dict[str, Dict[str, Any]]:
        """Fetch prices for list of symbols.

        Args:
            symbols: List of coin symbols (e.g., ["BTC", "ETH"])

        Returns:
            Dict mapping symbol to price data
        """
        # Check cache first
        if self._is_cache_valid():
            if self.verbose:
                print("Using cached prices", file=sys.stderr)
            return self._get_cached_prices(symbols)

        # Map symbols to CoinGecko IDs
        id_to_symbol = {}
        unknown = []

        for symbol in symbols:
            symbol = symbol.upper()
            coin_id = SYMBOL_TO_ID.get(symbol, symbol.lower())
            id_to_symbol[coin_id] = symbol

            if symbol not in SYMBOL_TO_ID:
                unknown.append(symbol)

        if unknown and self.verbose:
            print(f"Warning: Unknown symbols, trying lowercase: {unknown}", file=sys.stderr)

        # Fetch from API
        prices = self._fetch_from_api(list(id_to_symbol.keys()))

        # Map back to symbols
        result = {}
        for coin_id, data in prices.items():
            symbol = id_to_symbol.get(coin_id, coin_id.upper())
            result[symbol] = data

        # Update cache
        self._cache = result
        self._cache_time = time.time()
        self._save_cache()

        return result

    def _fetch_from_api(self, coin_ids: List[str]) -> Dict[str, Dict[str, Any]]:
        """Fetch prices from CoinGecko API."""
        results = {}

        # Batch into groups of 250 (API limit)
        batches = [coin_ids[i:i+250] for i in range(0, len(coin_ids), 250)]

        for batch in batches:
            try:
                if self.verbose:
                    print(f"Fetching prices for {len(batch)} coins...", file=sys.stderr)

                response = requests.get(
                    f"{self.COINGECKO_API}/coins/markets",
                    params={
                        "vs_currency": "usd",
                        "ids": ",".join(batch),
                        "order": "market_cap_desc",
                        "sparkline": "false",
                        "price_change_percentage": "24h,7d,30d"
                    },
                    timeout=15
                )
                response.raise_for_status()

                data = response.json()
                for coin in data:
                    results[coin["id"]] = {
                        "price": coin.get("current_price", 0),
                        "change_24h": coin.get("price_change_percentage_24h"),
                        "change_7d": coin.get("price_change_percentage_7d_in_currency"),
                        "change_30d": coin.get("price_change_percentage_30d_in_currency"),
                        "market_cap": coin.get("market_cap"),
                        "volume_24h": coin.get("total_volume"),
                        "last_updated": coin.get("last_updated")
                    }

                # Rate limit protection
                if len(batches) > 1:
                    time.sleep(0.5)

            except requests.exceptions.RequestException as e:
                if self.verbose:
                    print(f"API request failed: {e}", file=sys.stderr)
                # Try to load from cache
                return self._load_cached_prices()

        return results

    def _is_cache_valid(self) -> bool:
        """Check if cache is still valid."""
        if not self._cache:
            self._load_cache_file()
        return (time.time() - self._cache_time) < self.CACHE_TTL

    def _get_cached_prices(self, symbols: List[str]) -> Dict[str, Dict[str, Any]]:
        """Get prices from cache."""
        return {s: self._cache.get(s.upper(), {}) for s in symbols}

    def _load_cache_file(self) -> None:
        """Load cache from file."""
        try:
            if self.CACHE_FILE.exists():
                with open(self.CACHE_FILE, "r") as f:
                    data = json.load(f)
                self._cache = data.get("prices", {})
                self._cache_time = data.get("timestamp", 0)
        except Exception:
            self._cache = {}
            self._cache_time = 0

    def _save_cache(self) -> None:
        """Save cache to file."""
        try:
            with open(self.CACHE_FILE, "w") as f:
                json.dump({
                    "prices": self._cache,
                    "timestamp": self._cache_time
                }, f)
        except Exception:
            pass  # Cache save failure is non-fatal

    def _load_cached_prices(self) -> Dict[str, Dict[str, Any]]:
        """Load prices from cache file (fallback)."""
        self._load_cache_file()
        if self._cache:
            if self.verbose:
                print("Using stale cached prices as fallback", file=sys.stderr)
        return self._cache


def main():
    """CLI entry point for testing."""
    import argparse

    parser = argparse.ArgumentParser(description="Fetch cryptocurrency prices")
    parser.add_argument("symbols", nargs="+", help="Coin symbols (BTC, ETH, etc.)")
    parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
    args = parser.parse_args()

    fetcher = PriceFetcher(verbose=args.verbose)
    prices = fetcher.fetch_prices(args.symbols)
    print(json.dumps(prices, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Valuation Engine

Calculates portfolio valuations, allocations, and P&L.
Generates risk flags for concentration warnings.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import sys
from datetime import datetime
from typing import Optional, Dict, Any, List


class ValuationEngine:
    """Calculates portfolio valuations and analytics."""

    def __init__(self, verbose: bool = False):
        """Initialize engine.

        Args:
            verbose: Enable verbose output
        """
        self.verbose = verbose

    def calculate(
        self,
        portfolio: Dict[str, Any],
        prices: Dict[str, Dict[str, Any]],
        threshold: float = 25.0
    ) -> Dict[str, Any]:
        """Calculate portfolio valuations.

        Args:
            portfolio: Validated portfolio dict
            prices: Price data keyed by symbol
            threshold: Allocation warning threshold (percent)

        Returns:
            Valuation result with holdings and analytics
        """
        holdings = portfolio.get("holdings", [])
        categories = portfolio.get("categories", {})

        # Calculate individual holdings
        valued_holdings = []
        total_value = 0
        total_cost = 0
        has_cost_basis = False

        for holding in holdings:
            coin = holding.get("coin", "").upper()
            quantity = holding.get("quantity", 0)
            cost_basis = holding.get("cost_basis")

            # Get price data
            price_data = prices.get(coin, {})
            price = price_data.get("price", 0)

            # Calculate value
            value = quantity * price

            entry = {
                "coin": coin,
                "quantity": quantity,
                "price_usd": price,
                "value_usd": value,
                "change_24h_pct": price_data.get("change_24h"),
                "change_7d_pct": price_data.get("change_7d"),
                "change_30d_pct": price_data.get("change_30d"),
                "market_cap": price_data.get("market_cap"),
            }

            # Add cost basis and P&L if available
            if cost_basis is not None:
                has_cost_basis = True
                cost = quantity * cost_basis
                pnl = value - cost
                pnl_pct = ((price - cost_basis) / cost_basis * 100) if cost_basis > 0 else 0

                entry["cost_basis"] = cost_basis
                entry["total_cost"] = cost
                entry["unrealized_pnl"] = pnl
                entry["pnl_pct"] = pnl_pct

                total_cost += cost

            # Track category
            if coin in categories:
                entry["category"] = categories[coin]

            # Add wallets if tracked
            if holding.get("wallets"):
                entry["wallets"] = holding["wallets"]

            valued_holdings.append(entry)
            total_value += value

        # Calculate allocations
        for holding in valued_holdings:
            if total_value > 0:
                holding["allocation_pct"] = (holding["value_usd"] / total_value) * 100
            else:
                holding["allocation_pct"] = 0

        # Calculate totals
        total_change_24h = self._calculate_total_change(valued_holdings, "change_24h_pct", total_value)
        total_change_7d = self._calculate_total_change(valued_holdings, "change_7d_pct", total_value)

        # Generate risk flags
        risk_flags = self._generate_risk_flags(valued_holdings, threshold)

        # Calculate category allocation
        category_allocation = self._calculate_category_allocation(valued_holdings, total_value)

        # Build result
        result = {
            "portfolio_name": portfolio.get("name", "My Portfolio"),
            "total_value_usd": round(total_value, 2),
            "change_24h": {
                "amount": round(total_change_24h["amount"], 2) if total_change_24h else None,
                "percent": round(total_change_24h["percent"], 2) if total_change_24h else None
            },
            "change_7d": {
                "amount": round(total_change_7d["amount"], 2) if total_change_7d else None,
                "percent": round(total_change_7d["percent"], 2) if total_change_7d else None
            },
            "holdings": valued_holdings,
            "holdings_count": len(valued_holdings),
            "risk_flags": risk_flags,
            "allocation_by_category": category_allocation
        }

        # Add total P&L if cost basis available
        if has_cost_basis and total_cost > 0:
            total_pnl = total_value - total_cost
            total_pnl_pct = (total_pnl / total_cost) * 100

            result["total_cost"] = round(total_cost, 2)
            result["total_unrealized_pnl"] = round(total_pnl, 2)
            result["total_pnl_pct"] = round(total_pnl_pct, 2)

        return result

    def _calculate_total_change(
        self,
        holdings: List[Dict[str, Any]],
        change_key: str,
        total_value: float
    ) -> Optional[Dict[str, float]]:
        """Calculate weighted total change."""
        if total_value <= 0:
            return None

        weighted_change = 0
        valid_weight = 0

        for holding in holdings:
            change = holding.get(change_key)
            value = holding.get("value_usd", 0)

            if change is not None and value > 0:
                weighted_change += (value / total_value) * change
                valid_weight += value / total_value

        if valid_weight < 0.5:  # Less than 50% of portfolio has change data
            return None

        # Calculate dollar amount
        # If current value is V and it changed by P%, then previous value was V/(1 + P/100)
        if weighted_change != -100:
            previous_value = total_value / (1 + weighted_change / 100)
            change_amount = total_value - previous_value
        else:
            change_amount = -total_value

        return {
            "amount": change_amount,
            "percent": weighted_change
        }

    def _generate_risk_flags(
        self,
        holdings: List[Dict[str, Any]],
        threshold: float
    ) -> List[str]:
        """Generate risk flags for portfolio."""
        flags = []

        # Check concentration
        for holding in holdings:
            allocation = holding.get("allocation_pct", 0)
            coin = holding.get("coin", "")

            if allocation > threshold:
                flags.append(f"{coin} allocation ({allocation:.1f}%) exceeds {threshold:.0f}% threshold")

        # Check for single asset dominance
        if holdings and holdings[0].get("allocation_pct", 0) > 50:
            coin = holdings[0].get("coin", "")
            flags.append(f"{coin} represents majority of portfolio (>50%)")

        # Check for lack of diversification
        if len(holdings) < 3:
            flags.append("Portfolio has limited diversification (<3 assets)")

        return flags

    def _calculate_category_allocation(
        self,
        holdings: List[Dict[str, Any]],
        total_value: float
    ) -> Dict[str, float]:
        """Calculate allocation by category."""
        if total_value <= 0:
            return {}

        categories = {}
        uncategorized = 0

        for holding in holdings:
            category = holding.get("category", "Other")
            value = holding.get("value_usd", 0)

            if category not in categories:
                categories[category] = 0
            categories[category] += value

        # Convert to percentages
        return {
            cat: round((val / total_value) * 100, 1)
            for cat, val in sorted(categories.items(), key=lambda x: -x[1])
        }


def main():
    """CLI entry point for testing."""
    import json

    # Test with sample data
    portfolio = {
        "name": "Test Portfolio",
        "holdings": [
            {"coin": "BTC", "quantity": 0.5, "cost_basis": 50000},
            {"coin": "ETH", "quantity": 10, "cost_basis": 2500},
            {"coin": "SOL", "quantity": 100, "cost_basis": 100}
        ],
        "categories": {
            "BTC": "Layer 1",
            "ETH": "Layer 1",
            "SOL": "Layer 1"
        }
    }

    prices = {
        "BTC": {"price": 95000, "change_24h": 2.5, "change_7d": 5.0},
        "ETH": {"price": 3200, "change_24h": 1.8, "change_7d": 3.5},
        "SOL": {"price": 180, "change_24h": 4.2, "change_7d": 8.0}
    }

    engine = ValuationEngine(verbose=True)
    result = engine.calculate(portfolio, prices, threshold=25.0)
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()