#!/usr/bin/env python3 """ GENESIS REVENUE FEEDBACK LOOP ============================== Tracks revenue impact of tasks and generates financial performance summaries. Implements the feedback loop between task execution and revenue outcomes, ensuring Genesis can prioritize work that generates real AUD. Usage: from core.revenue_feedback_loop import record_task_revenue_impact, get_revenue_summary record_task_revenue_impact( task_id="RAI-007", revenue_impact="Widget upsell to $997/mo tier", outcome="success", amount_aud=997.0 ) summary = get_revenue_summary() # CLI: python3 revenue_feedback_loop.py --summary python3 revenue_feedback_loop.py --record RAI-007 "Widget sale" success 997.0 """ import json import sys import argparse from datetime import datetime, timezone from pathlib import Path from typing import Optional # All on E: drive (RULE 6) GENESIS_ROOT = Path("/mnt/e/genesis-system") LOOP_DIR = GENESIS_ROOT / "loop" REVENUE_LOG_PATH = LOOP_DIR / "revenue_log.jsonl" TASKS_JSON_PATH = LOOP_DIR / "tasks.json" VALID_OUTCOMES = {"success", "failure", "pending"} def _ensure_dirs(): """Ensure required directories exist.""" LOOP_DIR.mkdir(parents=True, exist_ok=True) def record_task_revenue_impact( task_id: str, revenue_impact: str, outcome: str, amount_aud: float = 0.0 ) -> dict: """ Record the revenue impact of a completed (or in-progress) task. Args: task_id: Genesis task ID (e.g. "RAI-007", "GEN-028") revenue_impact: Human-readable description of the revenue event outcome: "success" | "failure" | "pending" amount_aud: Dollar value tracked (AUD), default 0.0 Returns: The record dict that was appended to the log. """ _ensure_dirs() if outcome not in VALID_OUTCOMES: raise ValueError(f"outcome must be one of {VALID_OUTCOMES}, got: {outcome!r}") record = { "task_id": task_id, "revenue_impact": revenue_impact, "outcome": outcome, "amount_aud": float(amount_aud), "timestamp": datetime.now(timezone.utc).isoformat() } with open(REVENUE_LOG_PATH, "a", encoding="utf-8") as f: f.write(json.dumps(record) + "\n") return record def get_revenue_summary() -> dict: """ Read revenue_log.jsonl and return an aggregated performance summary. Returns: { total_tasks: int — unique task IDs recorded, revenue_generating: int — tasks with amount_aud > 0, total_aud_tracked: float — sum of all amount_aud, success_count: int, failure_count: int, pending_count: int, top_tasks: list — top 5 by amount_aud (desc) } """ _ensure_dirs() records = [] if REVENUE_LOG_PATH.exists(): with open(REVENUE_LOG_PATH, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: try: records.append(json.loads(line)) except json.JSONDecodeError: continue if not records: return { "total_tasks": 0, "revenue_generating": 0, "total_aud_tracked": 0.0, "success_count": 0, "failure_count": 0, "pending_count": 0, "top_tasks": [] } unique_task_ids = {r["task_id"] for r in records} revenue_generating = sum(1 for r in records if r.get("amount_aud", 0.0) > 0) total_aud = sum(r.get("amount_aud", 0.0) for r in records) outcome_counts = {"success": 0, "failure": 0, "pending": 0} for r in records: outcome = r.get("outcome", "pending") if outcome in outcome_counts: outcome_counts[outcome] += 1 sorted_records = sorted(records, key=lambda r: r.get("amount_aud", 0.0), reverse=True) top_tasks = sorted_records[:5] return { "total_tasks": len(unique_task_ids), "revenue_generating": revenue_generating, "total_aud_tracked": round(total_aud, 2), "success_count": outcome_counts["success"], "failure_count": outcome_counts["failure"], "pending_count": outcome_counts["pending"], "top_tasks": top_tasks } def update_tasks_with_revenue_field() -> dict: """ Read tasks.json and add `revenue_impact` field (default: "unknown") to any stories that are missing it. Writes updated JSON back in-place. Returns: {"stories_updated": int, "stories_already_had_field": int} """ if not TASKS_JSON_PATH.exists(): return {"error": f"tasks.json not found at {TASKS_JSON_PATH}"} with open(TASKS_JSON_PATH, "r", encoding="utf-8") as f: data = json.load(f) stories = data.get("stories", []) updated = 0 already_present = 0 for story in stories: if "revenue_impact" not in story: story["revenue_impact"] = "unknown" updated += 1 else: already_present += 1 # Stamp the update time data["updated_at"] = datetime.now(timezone.utc).isoformat() with open(TASKS_JSON_PATH, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) return {"stories_updated": updated, "stories_already_had_field": already_present} # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def _print_summary(summary: dict): print("\n=== GENESIS REVENUE FEEDBACK LOOP SUMMARY ===") print(f" Unique tasks tracked : {summary['total_tasks']}") print(f" Revenue-generating : {summary['revenue_generating']}") print(f" Total AUD tracked : ${summary['total_aud_tracked']:,.2f}") print(f" Outcomes — success: {summary['success_count']} failure: {summary['failure_count']} pending: {summary['pending_count']}") if summary["top_tasks"]: print("\n Top 5 highest-impact tasks:") for i, t in enumerate(summary["top_tasks"], 1): print(f" {i}. [{t['task_id']}] ${t['amount_aud']:,.2f} AUD — {t['revenue_impact'][:60]} ({t['outcome']})") else: print("\n No revenue records found yet.") print() def main(): parser = argparse.ArgumentParser( description="Genesis Revenue Feedback Loop — track and summarise task revenue impact" ) parser.add_argument("--summary", action="store_true", help="Print revenue summary") parser.add_argument( "--record", nargs=4, metavar=("TASK_ID", "REVENUE_IMPACT", "OUTCOME", "AMOUNT_AUD"), help="Record a revenue event: TASK_ID 'description' success|failure|pending 997.0" ) parser.add_argument( "--patch-tasks", action="store_true", help="Add missing revenue_impact field to all stories in tasks.json" ) args = parser.parse_args() if args.record: task_id, revenue_impact, outcome, amount_str = args.record try: amount = float(amount_str) except ValueError: print(f"ERROR: AMOUNT_AUD must be a number, got: {amount_str!r}") sys.exit(1) rec = record_task_revenue_impact(task_id, revenue_impact, outcome, amount) print(f"Recorded: {json.dumps(rec, indent=2)}") if args.patch_tasks: result = update_tasks_with_revenue_field() print(f"tasks.json patched: {result}") if args.summary or (not args.record and not args.patch_tasks): summary = get_revenue_summary() _print_summary(summary) if __name__ == "__main__": main()