""" metrics_service.py ================== Builds and caches BinaryTreeMetrics + BinaryLevelSnapshot for each DistributorID. Called: • Inline after purchase (lightweight incremental update). • By the rebuild_binary_tree_metrics Celery task (full rebuild, periodic). • By the binary analysis API (on-demand refresh). """ import logging from collections import defaultdict, deque from decimal import Decimal, ROUND_HALF_UP from django.db import transaction from apps.business.distributor.models import DistributorID from apps.business.binary_tree.models import BinaryTreeMetrics, BinaryLevelSnapshot from apps.business.binary_tree.services.binary_service import ( get_binary_left_right_volumes, get_binary_tree_depth, _children_ids, ) from apps.business.rewards.models import DistributorLevelProgress from apps.business.schemes.models import OpportunityBundleLevel logger = logging.getLogger(__name__) # ── Level resolution (same algorithm as level_service, using binary counts) ── def _build_cumulative_levels(levels): cumulative = 0 entries = [] for level in sorted(levels, key=lambda l: l.level_number): cumulative += int(getattr(level, 'required_positions', 0) or 0) entries.append({ 'level_number': level.level_number, 'required_positions': int(getattr(level, 'required_positions', 0) or 0), 'cumulative_required_positions': cumulative, }) return entries def _resolve_level_from_count(total_positions, cumulative_levels): highest_level = 0 current_threshold = 0 next_level = None next_required = None for entry in cumulative_levels: threshold = entry['cumulative_required_positions'] if total_positions >= threshold: highest_level = entry['level_number'] current_threshold = threshold else: next_level = entry['level_number'] next_required = threshold break if next_required is not None: span = max(1, next_required - current_threshold) filled = max(0, total_positions - current_threshold) progress_percent = Decimal(str(round(min(100.0, filled / span * 100), 1))) positions_to_next = max(0, next_required - total_positions) elif cumulative_levels: progress_percent = Decimal('100.0') positions_to_next = 0 else: progress_percent = Decimal('0.0') positions_to_next = 0 return { 'current_level': highest_level, 'next_level': next_level, 'next_level_required': next_required, 'positions_to_next': positions_to_next, 'progress_percent': progress_percent, } # ── Per-distributor metrics computation ────────────────────────────────────── def compute_metrics_for_distributor(distributor): """ Compute full BinaryTreeMetrics + BinaryLevelSnapshot for one distributor. Expensive: does full subtree traversal. Use build_metrics_for_many for bulk. """ dist_id = distributor.pk left_vol, right_vol = get_binary_left_right_volumes(dist_id) subtree_total = left_vol + right_vol depth = get_binary_tree_depth(dist_id) left_child_id, right_child_id = _children_ids(dist_id) direct_left = 1 if left_child_id else 0 direct_right = 1 if right_child_id else 0 if left_vol == right_vol: weak_leg = 'BALANCED' strong_leg = 'BALANCED' imbalance = Decimal('0') elif left_vol < right_vol: weak_leg = 'LEFT' strong_leg = 'RIGHT' imbalance = Decimal(str( round((right_vol - left_vol) / max(1, right_vol) * 100, 2) )) else: weak_leg = 'RIGHT' strong_leg = 'LEFT' imbalance = Decimal(str( round((left_vol - right_vol) / max(1, left_vol) * 100, 2) )) return { 'distributor': distributor, 'left_volume': left_vol, 'right_volume': right_vol, 'subtree_total': subtree_total, 'direct_left_count': direct_left, 'direct_right_count': direct_right, 'tree_depth': depth, 'weak_leg': weak_leg, 'strong_leg': strong_leg, 'imbalance_ratio': imbalance, } def upsert_metrics(distributor, metrics_data): """Persist BinaryTreeMetrics for one distributor.""" BinaryTreeMetrics.objects.update_or_create( distributor=distributor, defaults={ 'left_volume': metrics_data['left_volume'], 'right_volume': metrics_data['right_volume'], 'subtree_total': metrics_data['subtree_total'], 'direct_left_count': metrics_data['direct_left_count'], 'direct_right_count': metrics_data['direct_right_count'], 'tree_depth': metrics_data['tree_depth'], 'weak_leg': metrics_data['weak_leg'], 'strong_leg': metrics_data['strong_leg'], 'imbalance_ratio': metrics_data['imbalance_ratio'], }, ) def compute_and_persist_all(bundle_id=None, only_stale=False): """ O(n) bulk rebuild of BinaryTreeMetrics for all distributors. Previous approach: per-node BFS DB queries → O(n²). New approach: load entire tree into memory once, compute all metrics via a single bottom-up DFS traversal, then bulk-write with executemany/ bulk_create. Typical time for 100K nodes: seconds instead of hours. Returns dict with processed/skipped/errors counts. """ from django.utils import timezone from datetime import timedelta qs = DistributorID.objects.all() if bundle_id: qs = qs.filter(product__opportunity_bundle_id=bundle_id) # ── Load tree into memory ──────────────────────────────────────────────── rows = list(qs.values('id', 'binary_parent_distributor_id', 'binary_position')) # children[node_id] = {'LEFT': child_id_or_None, 'RIGHT': child_id_or_None} children: dict = {} all_ids: set = set() for row in rows: node_id = row['id'] all_ids.add(node_id) if node_id not in children: children[node_id] = {'LEFT': None, 'RIGHT': None} parent_id = row['binary_parent_distributor_id'] if parent_id and row['binary_position']: if parent_id not in children: children[parent_id] = {'LEFT': None, 'RIGHT': None} children[parent_id][row['binary_position']] = node_id # ── Single bottom-up DFS to compute subtree sizes + depths ─────────────── # size[n] = total nodes in subtree rooted at n (including n itself) # depth[n] = max depth from n (0 for a leaf) size: dict = {} depth: dict = {} # Find root nodes (no binary parent in this scope) parent_ids_set = { row['binary_parent_distributor_id'] for row in rows if row['binary_parent_distributor_id'] } roots = [nid for nid in all_ids if nid not in parent_ids_set] # Iterative post-order DFS visited: set = set() stack = list(roots) processing_order: list = [] # post-order (leaves first, roots last) while stack: node = stack[-1] if node not in visited: visited.add(node) ch = children.get(node, {}) left_c = ch.get('LEFT') right_c = ch.get('RIGHT') if left_c and left_c not in visited: stack.append(left_c) elif right_c and right_c not in visited: stack.append(right_c) else: stack.pop() processing_order.append(node) else: # Already visited — may still have right child unvisited stack.pop() ch = children.get(node, {}) right_c = ch.get('RIGHT') if right_c and right_c not in visited: visited.remove(node) # allow revisiting with right child stack.append(node) stack.append(right_c) else: processing_order.append(node) # Simpler alternative if DFS is complex — use Kahn's topo sort on the tree # (leaves have no children so they're processed first) if len(processing_order) < len(all_ids): # Fall back to Kahn's topological sort (handles all edge cases) processing_order = [] in_degree = {nid: 0 for nid in all_ids} # will count parent → child edges for nid in all_ids: ch = children.get(nid, {}) for child_id in (ch.get('LEFT'), ch.get('RIGHT')): if child_id and child_id in all_ids: in_degree[nid] += 1 # nid is a parent, child has parent # Actually for topo sort, we want: process nodes with no children first # i.e. nodes that no one else has as children # Compute out-degree instead: out_degree = defaultdict(int) rev_children = defaultdict(list) # child_id -> list of parents for nid in all_ids: ch = children.get(nid, {}) for child_id in (ch.get('LEFT'), ch.get('RIGHT')): if child_id and child_id in all_ids: out_degree[nid] += 1 rev_children[child_id].append(nid) q = deque(nid for nid in all_ids if out_degree[nid] == 0) # leaves while q: nid = q.popleft() processing_order.append(nid) for parent in rev_children[nid]: out_degree[parent] -= 1 if out_degree[parent] == 0: q.append(parent) # Compute sizes and depths in post-order (leaves first) for nid in processing_order: ch = children.get(nid, {}) left_c = ch.get('LEFT') right_c = ch.get('RIGHT') left_size = size.get(left_c, 0) if left_c else 0 right_size = size.get(right_c, 0) if right_c else 0 size[nid] = 1 + left_size + right_size left_depth = depth.get(left_c, -1) if left_c else -1 right_depth = depth.get(right_c, -1) if right_c else -1 depth[nid] = 1 + max(left_depth, right_depth) if (left_c or right_c) else 0 # ── Build ORM objects for bulk write ───────────────────────────────────── metrics_objects = [] for nid in all_ids: ch = children.get(nid, {}) left_c = ch.get('LEFT') right_c = ch.get('RIGHT') left_vol = size.get(left_c, 0) if left_c else 0 right_vol = size.get(right_c, 0) if right_c else 0 subtree_total = left_vol + right_vol node_depth = depth.get(nid, 0) direct_left = 1 if left_c else 0 direct_right = 1 if right_c else 0 if left_vol == right_vol: weak_leg = 'BALANCED' strong_leg = 'BALANCED' imbalance = Decimal('0.00') elif left_vol < right_vol: weak_leg = 'LEFT' strong_leg = 'RIGHT' imbalance = Decimal(str( round((right_vol - left_vol) / max(1, right_vol) * 100, 2) )) else: weak_leg = 'RIGHT' strong_leg = 'LEFT' imbalance = Decimal(str( round((left_vol - right_vol) / max(1, left_vol) * 100, 2) )) metrics_objects.append(BinaryTreeMetrics( distributor_id=nid, left_volume=left_vol, right_volume=right_vol, subtree_total=subtree_total, direct_left_count=direct_left, direct_right_count=direct_right, tree_depth=node_depth, weak_leg=weak_leg, strong_leg=strong_leg, imbalance_ratio=imbalance, )) # ── Bulk-write via ORM (handles UUID serialization correctly) ───────────── UPDATE_FIELDS = [ 'left_volume', 'right_volume', 'subtree_total', 'direct_left_count', 'direct_right_count', 'tree_depth', 'weak_leg', 'strong_leg', 'imbalance_ratio', ] BATCH = 2000 errors = 0 processed = 0 with transaction.atomic(): for i in range(0, len(metrics_objects), BATCH): chunk = metrics_objects[i: i + BATCH] try: BinaryTreeMetrics.objects.bulk_create( chunk, update_conflicts=True, update_fields=UPDATE_FIELDS, unique_fields=['distributor'], ) processed += len(chunk) except Exception as exc: logger.error("[METRICS BULK] Batch %d failed: %s", i // BATCH, exc) errors += len(chunk) logger.info("[METRICS] Bulk-wrote %d metrics records.", processed) return {'processed': processed, 'skipped': 0, 'errors': errors} # ── Binary-based level snapshot ─────────────────────────────────────────────── def compute_binary_level_snapshot(distributor): """ Calculate and persist BinaryLevelSnapshot using the binary subtree count. Also mirrors the result into DistributorLevelProgress so existing APIs (rank_progress, dashboard) automatically use binary counts. """ bundle_id = getattr(getattr(distributor, 'product', None), 'opportunity_bundle_id', None) if not bundle_id: return None # Get or compute binary metrics try: metrics = BinaryTreeMetrics.objects.get(distributor=distributor) subtree_total = metrics.subtree_total left_vol = metrics.left_volume right_vol = metrics.right_volume except BinaryTreeMetrics.DoesNotExist: data = compute_metrics_for_distributor(distributor) upsert_metrics(distributor, data) subtree_total = data['subtree_total'] left_vol = data['left_volume'] right_vol = data['right_volume'] # Build cumulative level thresholds from OpportunityBundleLevel levels = list( OpportunityBundleLevel.objects.filter( opportunity_bundle_id=bundle_id ).order_by('level_number') ) cumulative_levels = _build_cumulative_levels(levels) level_data = _resolve_level_from_count(subtree_total, cumulative_levels) # Fetch existing snapshot to preserve last_completed_level existing = BinaryLevelSnapshot.objects.filter(distributor=distributor).first() last_completed_level = 0 if existing: last_completed_level = min(existing.last_completed_level, level_data['current_level']) snapshot, _ = BinaryLevelSnapshot.objects.update_or_create( distributor=distributor, defaults={ 'subtree_total': subtree_total, 'left_volume': left_vol, 'right_volume': right_vol, 'current_level': level_data['current_level'], 'last_completed_level': last_completed_level, 'next_level': level_data['next_level'], 'next_level_required': level_data['next_level_required'], 'positions_to_next': level_data['positions_to_next'], 'progress_percent': level_data['progress_percent'], }, ) # Mirror into DistributorLevelProgress so existing APIs stay consistent DistributorLevelProgress.objects.update_or_create( distributor=distributor, defaults={ 'current_level': level_data['current_level'], 'last_completed_level': last_completed_level, 'total_positions_after': subtree_total, }, ) return snapshot def get_binary_dashboard_data(distributor): """ Return the full binary dashboard dict for one distributor. Used by both web and admin APIs. Response shape matches the frontend BinaryDashboard type: { distributor_id, distributor_code, binary_parent_id, binary_position, hint, metrics: { BinaryTreeMetrics fields }, level_snapshot: { BinaryLevelSnapshot fields } | null, } """ # Always recompute fresh — catches stale cached metrics when children are added data = compute_metrics_for_distributor(distributor) upsert_metrics(distributor, data) metrics = BinaryTreeMetrics.objects.get(distributor_id=distributor.pk) snapshot = compute_binary_level_snapshot(distributor) # Count active direct sponsor referrals direct_sponsor_count = DistributorID.objects.filter( sponsor_distributor=distributor, is_active=True ).count() # Build sponsor recruits list showing where each recruit landed in the binary tree left_child_id, right_child_id = _children_ids(distributor.pk) direct_binary_set = {str(cid) for cid in [left_child_id, right_child_id] if cid} sponsor_recruits_qs = DistributorID.objects.filter( sponsor_distributor=distributor ).order_by('global_position').select_related('user').values( 'id', 'distributor_code', 'global_position', 'binary_position', 'user__first_name', 'user__last_name', 'user__referral_code', ) sponsor_recruits = [ { 'code': r['distributor_code'], 'global_position': r['global_position'], 'binary_side': r['binary_position'], 'is_direct_binary_child': str(r['id']) in direct_binary_set, 'owner_full_name': ( f"{r.get('user__first_name') or ''} {r.get('user__last_name') or ''}".strip() or 'Member' ), 'owner_referral_code': r.get('user__referral_code'), } for r in sponsor_recruits_qs ] metrics_dict = { 'distributor_id': str(distributor.id), 'distributor_code': distributor.distributor_code, 'left_volume': metrics.left_volume, 'right_volume': metrics.right_volume, 'subtree_total': metrics.subtree_total, 'direct_left_count': metrics.direct_left_count, 'direct_right_count': metrics.direct_right_count, 'tree_depth': metrics.tree_depth, 'weak_leg': metrics.weak_leg or 'BALANCED', 'strong_leg': metrics.strong_leg or 'BALANCED', 'imbalance_ratio': str(metrics.imbalance_ratio), 'updated_at': metrics.last_rebuilt_at.isoformat(), } snapshot_dict = None if snapshot: snapshot_dict = { 'distributor_id': str(distributor.id), 'subtree_total': snapshot.subtree_total, 'left_volume': snapshot.left_volume, 'right_volume': snapshot.right_volume, 'current_level': snapshot.current_level, 'last_completed_level': snapshot.last_completed_level, 'next_level_threshold': snapshot.next_level_required, 'progress_percent': str(snapshot.progress_percent), 'positions_to_next': snapshot.positions_to_next, 'next_level': snapshot.next_level, 'updated_at': snapshot.calculated_at.isoformat() if snapshot.calculated_at else None, } # Build full bundle level ladder for the frontend level progress UI bundle_levels = [] bundle_id = getattr(getattr(distributor, 'product', None), 'opportunity_bundle_id', None) if bundle_id: levels_qs = OpportunityBundleLevel.objects.filter( opportunity_bundle_id=bundle_id ).order_by('level_number').values('level_number', 'required_positions') cumulative = 0 for lvl in levels_qs: cumulative += int(lvl['required_positions'] or 0) bundle_levels.append({ 'level': lvl['level_number'], 'positions': int(lvl['required_positions'] or 0), 'cumulative': cumulative, }) return { 'distributor_id': str(distributor.id), 'distributor_code': distributor.distributor_code, 'global_position': distributor.global_position, 'binary_parent_id': str(distributor.binary_parent_distributor_id) if distributor.binary_parent_distributor_id else None, 'binary_position': distributor.binary_position or None, 'direct_sponsor_referrals': direct_sponsor_count, 'sponsor_recruits': sponsor_recruits, 'hint': _generate_hint(metrics, snapshot, direct_sponsor_count), 'metrics': metrics_dict, 'level_snapshot': snapshot_dict, 'bundle_levels': bundle_levels, } def _generate_hint(metrics, snapshot, direct_sponsor_referrals=0): """Generate an actionable text hint based on current binary tree state.""" # No binary children yet (leaf node in global BFS tree) if metrics.subtree_total == 0: if direct_sponsor_referrals > 0: return ( f"You have {direct_sponsor_referrals} direct recruit(s) in the sponsor network. " "In the Global Assisted Binary system, new members are placed at the next available " "BFS position globally — so your recruits may be placed outside your direct binary " "subtree. Your LEFT and RIGHT slots will be filled as the company grows the tree " "to your position." ) return "You are placed in the binary network. Your LEFT and RIGHT slots will be filled as the global tree expands to your position." if metrics.direct_left_count == 0: return "Fill your LEFT slot first — invite a direct member to the left." if metrics.direct_right_count == 0: return "Fill your RIGHT slot — invite a direct member to the right side." if metrics.weak_leg == 'LEFT': missing = metrics.right_volume - metrics.left_volume return f"Focus on LEFT leg — you need {missing} more position(s) to balance." if metrics.weak_leg == 'RIGHT': missing = metrics.left_volume - metrics.right_volume return f"Focus on RIGHT leg — you need {missing} more position(s) to balance." if snapshot and snapshot.positions_to_next > 0: return ( f"Great balance! Add {snapshot.positions_to_next} more " f"position(s) to reach Level {snapshot.next_level}." ) return "Your binary tree is well balanced. Keep growing both legs!"