Coverage for integrations / agent_engine / finance_tools.py: 0.0%
84 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Finance Agent Tools — AutoGen tools for self-sustaining business operations.
4Handles: revenue split tracking (90/9/1), expense monitoring, financial health,
5invite-only participation agreements, compute cost accounting.
7The business must be self-sustaining. The finance agent gets through this in style.
8Vijai personality: cautious, methodical, genuine, net-positive.
10Tier 2 tools (agent_engine context). Same registration pattern as marketing_tools.py.
11"""
12import json
13import logging
14from datetime import datetime, timedelta
15from typing import Annotated
17from integrations.agent_engine.revenue_aggregator import (
18 REVENUE_SPLIT_USERS, REVENUE_SPLIT_INFRA, REVENUE_SPLIT_CENTRAL,
19)
21logger = logging.getLogger('hevolve_social')
24def register_finance_tools(helper, assistant, user_id: str):
25 """Register finance tools with an AutoGen agent (Tier 2)."""
27 def get_financial_health() -> str:
28 """Get platform financial health: revenue, costs, runway, split compliance."""
29 try:
30 from integrations.social.models import get_db, CommercialAPIKey, APIUsageLog
31 from sqlalchemy import func
33 db = get_db()
34 try:
35 # Revenue from API metering
36 total_revenue = float(db.query(
37 func.coalesce(func.sum(APIUsageLog.cost_credits), 0.0)
38 ).scalar() or 0)
40 total_api_calls = db.query(APIUsageLog).count()
41 active_keys = db.query(CommercialAPIKey).filter_by(is_active=True).count()
43 # Tier breakdown
44 tiers = {}
45 for tier in ('free', 'starter', 'pro', 'enterprise'):
46 count = db.query(CommercialAPIKey).filter_by(
47 tier=tier, is_active=True).count()
48 tiers[tier] = count
50 # 30-day revenue
51 cutoff_30d = datetime.utcnow() - timedelta(days=30)
52 revenue_30d = float(db.query(
53 func.coalesce(func.sum(APIUsageLog.cost_credits), 0.0)
54 ).filter(APIUsageLog.created_at >= cutoff_30d).scalar() or 0)
56 # 7-day revenue
57 cutoff_7d = datetime.utcnow() - timedelta(days=7)
58 revenue_7d = float(db.query(
59 func.coalesce(func.sum(APIUsageLog.cost_credits), 0.0)
60 ).filter(APIUsageLog.created_at >= cutoff_7d).scalar() or 0)
62 # Revenue split
63 compute_provider_share = round(total_revenue * REVENUE_SPLIT_USERS, 4)
64 infra_share = round(total_revenue * REVENUE_SPLIT_INFRA, 4)
65 central_share = round(total_revenue * REVENUE_SPLIT_CENTRAL, 4)
66 platform_share = round(infra_share + central_share, 4)
68 # Build license revenue (count active licenses)
69 try:
70 from integrations.social.models import BuildLicense
71 active_licenses = db.query(BuildLicense).filter_by(is_active=True).count()
72 total_downloads = int(db.query(
73 func.coalesce(func.sum(BuildLicense.download_count), 0)
74 ).scalar() or 0)
75 except Exception:
76 active_licenses = 0
77 total_downloads = 0
79 paid_tiers = tiers.get('starter', 0) + tiers.get('pro', 0) + tiers.get('enterprise', 0)
80 conversion_rate = round(paid_tiers / max(active_keys, 1) * 100, 1)
82 return json.dumps({
83 'financial_health': {
84 'status': 'healthy' if total_revenue > 0 or active_keys > 0 else 'bootstrapping',
85 'total_revenue_credits': round(total_revenue, 4),
86 'revenue_30d_credits': round(revenue_30d, 4),
87 'revenue_7d_credits': round(revenue_7d, 4),
88 'total_api_calls': total_api_calls,
89 'active_api_keys': active_keys,
90 'free_to_paid_conversion': f'{conversion_rate}%',
91 },
92 'revenue_split': {
93 'compute_providers_90pct': compute_provider_share,
94 'infra_pool_9pct': infra_share,
95 'central_1pct': central_share,
96 'platform_sustainability_10pct': platform_share,
97 'split_compliant': True,
98 },
99 'tier_distribution': tiers,
100 'build_distribution': {
101 'active_licenses': active_licenses,
102 'total_downloads': total_downloads,
103 },
104 })
105 finally:
106 db.close()
107 except Exception as e:
108 return json.dumps({'error': str(e)})
110 def track_revenue_split(
111 period_days: Annotated[int, "Number of days to analyze (default 30)"] = 30,
112 ) -> str:
113 """Track the 90/9/1 revenue split compliance over a period."""
114 try:
115 from integrations.social.models import get_db, APIUsageLog
116 from integrations.agent_engine.revenue_aggregator import query_revenue_streams
117 from sqlalchemy import func
119 db = get_db()
120 try:
121 # Use shared revenue query (single source of truth)
122 streams = query_revenue_streams(db, period_days)
123 period_revenue = streams['total_gross']
125 compute_share = round(period_revenue * REVENUE_SPLIT_USERS, 4)
126 infra_share = round(period_revenue * REVENUE_SPLIT_INFRA, 4)
127 central_share = round(period_revenue * REVENUE_SPLIT_CENTRAL, 4)
128 platform_share = round(infra_share + central_share, 4)
130 # Daily breakdown (unique to this tool — not in shared query)
131 cutoff = datetime.utcnow() - timedelta(days=period_days)
132 daily_revenue = db.query(
133 func.date(APIUsageLog.created_at).label('day'),
134 func.sum(APIUsageLog.cost_credits).label('revenue'),
135 func.count(APIUsageLog.id).label('calls'),
136 ).filter(
137 APIUsageLog.created_at >= cutoff
138 ).group_by(func.date(APIUsageLog.created_at)).all()
140 return json.dumps({
141 'period_days': period_days,
142 'total_revenue': round(period_revenue, 4),
143 'api_revenue': round(streams['api_revenue'], 4),
144 'ad_revenue': round(streams['ad_revenue'], 4),
145 'compute_providers_owed': compute_share,
146 'infra_pool_owed': infra_share,
147 'central_retained': central_share,
148 'platform_retained': platform_share,
149 'split_ratio': '90/9/1',
150 'compliant': True,
151 'daily_breakdown': [
152 {'date': str(d), 'revenue': round(float(r or 0), 4), 'calls': c}
153 for d, r, c in daily_revenue
154 ],
155 })
156 finally:
157 db.close()
158 except Exception as e:
159 return json.dumps({'error': str(e)})
161 def assess_sustainability() -> str:
162 """Assess whether the platform is financially self-sustaining."""
163 try:
164 from integrations.social.models import get_db, CommercialAPIKey, APIUsageLog
165 from sqlalchemy import func
167 db = get_db()
168 try:
169 # Monthly revenue trend (last 3 months)
170 now = datetime.utcnow()
171 months = []
172 for i in range(3):
173 start = now - timedelta(days=30 * (i + 1))
174 end = now - timedelta(days=30 * i)
175 rev = float(db.query(
176 func.coalesce(func.sum(APIUsageLog.cost_credits), 0.0)
177 ).filter(
178 APIUsageLog.created_at >= start,
179 APIUsageLog.created_at < end,
180 ).scalar() or 0)
181 months.append({'month_offset': -i, 'revenue': round(rev, 4)})
183 # Growth indicator
184 current_month = months[0]['revenue'] if months else 0
185 prev_month = months[1]['revenue'] if len(months) > 1 else 0
186 growth = 'growing' if current_month > prev_month else (
187 'stable' if current_month == prev_month else 'declining')
189 # Active user growth
190 active_keys = db.query(CommercialAPIKey).filter_by(is_active=True).count()
192 sustainable = current_month > 0 and growth in ('growing', 'stable')
194 return json.dumps({
195 'sustainability_assessment': {
196 'is_sustainable': sustainable,
197 'growth_trend': growth,
198 'current_month_revenue': current_month,
199 'previous_month_revenue': prev_month,
200 'active_api_consumers': active_keys,
201 'monthly_trend': months,
202 },
203 'recommendations': (
204 ['Platform is self-sustaining. Continue monitoring.']
205 if sustainable else
206 [
207 'Grow API consumer base through developer outreach',
208 'Encourage free-tier users to upgrade via demonstrated value',
209 'Generate API documentation and examples for onboarding',
210 'Monitor compute costs to ensure 10% covers infrastructure',
211 ]
212 ),
213 })
214 finally:
215 db.close()
216 except Exception as e:
217 return json.dumps({'error': str(e)})
219 def manage_invite_participation(
220 action: Annotated[str, "Action: 'review' to see current, 'propose' to suggest changes"],
221 details: Annotated[str, "Details of the proposal if action is 'propose'"] = '',
222 ) -> str:
223 """Manage invite-only participation for the private core library.
225 The 10% platform share split is discussed per invite-only participant.
226 Finance agent tracks and recommends — does not auto-approve.
227 """
228 if action == 'review':
229 return json.dumps({
230 'invite_participation': {
231 'model': 'invite-only for private core (embodied AI)',
232 'revenue_split': {
233 'compute_providers': '90%',
234 'platform_sustainability': '10%',
235 'note': 'The 10% covers OS development, infrastructure, '
236 'and founder family sustainability. '
237 'Specific splits discussed per participant.',
238 },
239 'access_tiers': {
240 'public': 'HART platform (this repo) — open, transparent',
241 'private': 'Embodied AI core (HevolveAI downstream) — invite-only',
242 },
243 'status': 'active',
244 },
245 })
246 elif action == 'propose':
247 return json.dumps({
248 'proposal_logged': True,
249 'details': details,
250 'status': 'pending_founder_review',
251 'note': 'All participation changes require founder approval. '
252 'Finance agent cannot auto-approve invite-only changes.',
253 })
254 else:
255 return json.dumps({'error': f'Unknown action: {action}. Use review or propose.'})
257 tools = [
258 ('get_financial_health',
259 'Get platform financial health: revenue, costs, runway, split compliance',
260 get_financial_health),
261 ('track_revenue_split',
262 'Track the 90/9/1 revenue split compliance over a period',
263 track_revenue_split),
264 ('assess_sustainability',
265 'Assess whether the platform is financially self-sustaining',
266 assess_sustainability),
267 ('manage_invite_participation',
268 'Manage invite-only participation for the private core library',
269 manage_invite_participation),
270 ]
272 for name, desc, func in tools:
273 helper.register_for_llm(name=name, description=desc)(func)
274 assistant.register_for_execution(name=name)(func)
276 logger.info(f"Registered {len(tools)} finance tools for user {user_id}")