fix: [BOUNTY] Trader: Trading portfolio auto balance algorithm, t

fix_issue_138.py

@@ -0,0 +1,3 @@
+```json
+{
+  "solution_code": "# python/trader/portfolio_balancer.py\n\"\"\"\nTrading Portfolio Auto-Balance Algorithm\nUses Deep (associative data storage) for asset management.\n\nIssue #138: Auto-balance portfolio based on target allocation percentages.\n\"\"\"\n\nfrom __future__ import annotations\nfrom dataclasses import dataclass, field\nfrom typing import Dict, List, Optional, Tuple\nimport json\nimport os\nimport time\n\n\n# ---------------------------------------------------------------------------\n# Minimal Deep-style associative storage\n# ---------------------------------------------------------------------------\n\nclass Deep:\n    \"\"\"\n    Lightweight associative (graph-based) storage that mimics the\n    LinksP latform Deep concept.\n\n    Every link has the shape  (id, type_id, source_id, target_id, value).\n    We store links in a flat dict keyed by id and provide simple helpers\n    to attach typed values so the rest of the code can reason about\n    portfolio data in an associative way.\n    \"\"\"\n\n    def __init__(self, path: str = \"portfolio.deep.json\"):\n        self._path = path\n        self._links: Dict[int, dict] = {}\n        self._next_id: int = 1\n        self._load()\n\n    # ---- persistence -------------------------------------------------------\n\n    def _load(self):\n        if os.path.exists(self._path):\n            try:\n                with open(self._path, \"r\") as fh:\n                    data = json.load(fh)\n                self._links = {int(k): v for k, v in data.get(\"links\", {}).items()}\n                self._next_id = data.get(\"next_id\", 1)\n            except (json.JSONDecodeError, KeyError):\n                self._links = {}\n                self._next_id = 1\n\n    def save(self):\n        with open(self._path, \"w\") as fh:\n            json.dump({\"links\": self._links, \"next_id\": self._next_id}, fh, indent=2)\n\n    # ---- core CRUD ---------------------------------------------------------\n\n    def create(self, type_id: int = 0, source_id: int = 0,\n               target_id: int = 0, value=None) -> int:\n        link_id = self._next_id\n        self._next_id += 1\n        self._links[link_id] = {\n            \"id\": link_id,\n            \"type\": type_id,\n            \"source\": source_id,\n            \"target\": target_id,\n            \"value\": value,\n            \"ts\": time.time(),\n        }\n        return link_id\n\n    def read(self, link_id: int) -> Optional[dict]:\n        return self._links.get(link_id)\n\n    def update(self, link_id: int, **kwargs) -> bool:\n        if link_id not in self._links:\n            return False\n        self._links[link_id].update(kwargs)\n        self._links[link_id][\"ts\"] = time.time()\n        return True\n\n    def delete(self, link_id: int) -> bool:\n        return self._links.pop(link_id, None) is not None\n\n    def query(self, **kwargs) -> List[dict]:\n        \"\"\"Return all links whose fields match the given keyword filters.\"\"\"\n        results = []\n        for link in self._links.values():\n            if all(link.get(k) == v for k, v in kwargs.items()):\n                results.append(link)\n        return results\n\n    # ---- named-type helpers ------------------------------------------------\n\n    # We reserve a few well-known type_ids so we don't need a separate registry.\n    TYPE_ASSET          = 1   # root node for an asset\n    TYPE_ALLOCATION     = 2   # target allocation link (asset -> pct value)\n    TYPE_HOLDING        = 3   # current holding link   (asset -> qty value)\n    TYPE_PRICE          = 4   # price link             (asset -> price value)\n    TYPE_REBALANCE_LOG  = 5   # audit log entry\n\n    def ensure_asset(self, symbol: str) -> int:\n        \"\"\"\n        Return the link-id of the asset node with the given symbol,\n        creating it if it does not exist.\n        \"\"\"\n        existing = self.query(type=self.TYPE_ASSET, value=symbol)\n        if existing:\n            return existing[0][\"id\"]\n        return self.create(type_id=self.TYPE_ASSET, value=symbol)\n\n    def set_allocation(self, asset_id: int, pct: float) -> int:\n        existing = self.query(type=self.TYPE_ALLOCATION, source=asset_id)\n        if existing:\n            lid = existing[0][\"id\"]\n            self.update(lid, value=pct)\n            return lid\n        return self.create(type_id=self.TYPE_ALLOCATION, source_id=asset_id, value=pct)\n\n    def get_allocation(self, asset_id: int) -> float:\n        rows = self.query(type=self.TYPE_ALLOCATION, source=asset_id)\n        return rows[0][\"value\"] if rows else 0.0\n\n    def set_holding(self, asset_id: int, qty: float) -> int:\n        existing = self.query(type=self.TYPE_HOLDING, source=asset_id)\n        if existing:\n            lid = existing[0][\"id\"]\n            self.update(lid, value=qty)\n            return lid\n        return self.create(type_id=self.TYPE_HOLDING, source_id=asset_id, value=qty)\n\n    def get_holding(self, asset_id: int) -> float:\n        rows = self.query(type=self.TYPE_HOLDING, source=asset_id)\n        return rows[0][\"value\"] if rows else 0.0\n\n    def set_price(self, asset_id: int, price: float) -> int:\n        existing = self.query(type=self.TYPE_PRICE, source=asset_id)\n        if existing:\n            lid = existing[0][\"id\"]\n            self.update(lid, value=price)\n            return lid\n        return self.create(type_id=self.TYPE_PRICE, source_id=asset_id, value=price)\n\n    def get_price(self, asset_id: int) -> float:\n        rows = self.query(type=self.TYPE_PRICE, source=asset_id)\n        return rows[0][\"value\"] if rows else 0.0\n\n    def all_assets(self) -> List[dict]:\n        return self.query(type=self.TYPE_ASSET)\n\n\n# ---------------------------------------------------------------------------\n# Domain objects\n# ---------------------------------------------------------------------------\n\n@dataclass\nclass AssetInfo:\n    symbol: str\n    target_pct: float   # 0-100\n    holding_qty: float  # units held\n    price: float        # price per unit in base currency\n\n    @property\n    def market_value(self) -> float:\n        return self.holding_qty * self.price\n\n\n@dataclass\nclass RebalanceOrder:\n    symbol: str\n    side: str           # 'BUY' or 'SELL'\n    qty: float          # units to trade\n    estimated_value: float\n\n    def __str__(self) -> str:\n        return (\n            f\"{self.side:4s}  {self.symbol:10s}  qty={self.qty:.6f}  \"\n            f\"est_value={self.estimated_value:.2f}\"\n        )\n\n\n# ---------------------------------------------------------------------------\n# Portfolio Balancer\n# ---------------------------------------------------------------------------\n\nclass PortfolioBalancer:\n    \"\"\"\n    Manages a portfolio whose target weights are stored in a Deep instance.\n\n    Workflow\n    --------\n    1. Define target allocations  (set_target_allocation)\n    2. Update current prices       (update_price)\n    3. Update current holdings     (update_holding)\n    4. Call rebalance()            -> list of RebalanceOrder\n    5. After executing orders, call commit_rebalance(orders) to persist\n       the new holding quantities.\n    \"\"\"\n\n    def __init__(self, deep: Deep, tolerance_pct: float = 1.0):\n        \"\"\"\n        Parameters\n        ----------\n        deep          : Deep storage instance\n        tolerance_pct : ignore drift smaller than this many percentage-points\n                        (avoids churning on tiny imbalances)\n        \"\"\"\n        self._deep = deep\n        self._tolerance = tolerance_pct  # percentage points\n\n    # ------------------------------------------------------------------ setup\n\n    def set_target_allocation(self, allocations: Dict[str, float]):\n        \"\"\"\n        Define the desired portfolio weights.\n\n        Parameters\n        ----------\n        allocations : dict mapping asset symbol -> target percentage (0-100)\n                      Percentages must sum to 100 (±0.01 rounding tolerance).\n\n        Example\n        -------\n        balancer.set_target_allocation({\n            'GOLD': 25.0,\n            'USD':  25.0,\n            'TCSG': 50.0,\n        })\n        \"\"\"\n        total = sum(allocations.values())\n        if abs(total - 100.0) > 0.01:\n            raise ValueError(\n                f\"Allocations must sum to 100 %, got {total:.4f} %\"\n            )\n        for symbol, pct in allocations.items():\n            if pct < 0:\n                raise ValueError(f\"Allocation for {symbol} cannot be negative.\")\n            asset_id = self._deep.ensure_asset(symbol)\n            self._deep.set_allocation(asset_id, pct)\n        self._deep.save()\n\n    def update_price(self, symbol: str, price: float):\n        \"\"\"Store the latest market price for an asset.\"\"\"\n        if price <= 0:\n            raise ValueError(f\"Price for {symbol} must be positive, got {price}.\")\n        asset_id = self._deep.ensure_asset(symbol)\n        self._deep.set_price(asset_id, price)\n        self._deep.save()\n\n    def update_holding(self, symbol: str, qty: float):\n        \"\"\"Store the current number of units held for an asset.\"\"\"\n        if qty < 0:\n            raise ValueError(f\"Holding quantity for {symbol} cannot be negative.\")\n        asset_id = self._deep.ensure_asset(symbol)\n        self._deep.set_holding(asset_id, qty)\n        self._deep.save()\n\n    # --------------------------------------------------------------- read state\n\n    def _get_asset_infos(self) -> List[AssetInfo]:\n        infos = []\n        for link in self._deep.all_assets():\n            aid = link[\"id\"]\n            sym = link[\"value\"]\n            infos.append(AssetInfo(\n                symbol=sym,\n                target_pct=self._deep.get_allocation(aid),\n                holding_qty=self._deep.get_holding(aid),\n                price=self._deep.get_price(aid),\n            ))\n        return infos\n\n    def portfolio_value(self) -> float:\n        \"\"\"Total current market value of the whole portfolio.\"\"\"\n