From b7eb6473cc626846da5a4f4f0992bea459babf6b Mon Sep 17 00:00:00 2001 From: moon <152454724+pabloDarkmoon24@users.noreply.github.com> Date: Mon, 23 Mar 2026 17:11:29 -0500 Subject: [PATCH] fix: solution for issue #138 --- fix_issue_138.py | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 fix_issue_138.py diff --git a/fix_issue_138.py b/fix_issue_138.py new file mode 100644 index 00000000..528f6992 --- /dev/null +++ b/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 \ No newline at end of file