"""Smart Coaching - Hybrid recommendation engine.
Rules-based thresholds for instant alerts + template-based narrative generation
for contextual 'Next Move' recommendations. Cached with 5-minute TTL.
Data sources: existing analytics routes (/api/analytics/*)
"""
import logging
from typing import Any
import requests as http_requests
from flask import Blueprint, current_app, jsonify, request
from app.routes.api_proxy import require_auth_json
from app.utils.csrf import require_csrf
logger = logging.getLogger(__name__)
coaching_bp = Blueprint('coaching', __name__, url_prefix='/api/coaching')
# ── In-memory cache (5-minute TTL) ──────────────────────────────────
_cache: dict[str, Any] = {}
_CACHE_TTL = 300 # seconds
def _cache_get(key: str) -> Any:
import time
if key in _cache:
value, timestamp = _cache[key]
if time.time() - timestamp < _CACHE_TTL:
return value
del _cache[key]
return None
def _cache_set(key: str, value: Any) -> None:
import time
_cache[key] = (value, time.time())
# ── Analytics data fetcher ───────────────────────────────────────────
def _fetch_analytics(endpoint: str, token: str | None = None) -> dict | None:
"""Fetch data from an existing analytics endpoint."""
headers = {}
if token:
headers['Authorization'] = f'Bearer {token}'
try:
base_url = request.host_url.rstrip('/')
resp = http_requests.get(f'{base_url}{endpoint}', headers=headers, timeout=5)
if resp.status_code == 200:
return resp.json()
except Exception as e:
logger.warning(f'Failed to fetch {endpoint}: {e}')
return None
# ── Rules Engine ─────────────────────────────────────────────────────
def evaluate_rules(token: str | None = None) -> list[dict]:
"""Evaluate rules-based thresholds against analytics data."""
insights: list[dict] = []
# Fetch analytics data
market_data = _fetch_analytics('/api/analytics/multi-market/overview', token)
forecast_data = _fetch_analytics('/api/analytics/forecast/pipeline', token)
goals_data = _fetch_analytics('/api/analytics/goals', token)
scale_data = _fetch_analytics('/api/analytics/scale-optimization/overview', token)
intelligence_data = _fetch_analytics('/api/analytics/strategic-intelligence/overview', token)
# Rule 1: Market underperformance (< 80% of target)
if market_data and isinstance(market_data, dict):
markets = market_data.get('markets', [])
if isinstance(markets, list):
for market in markets:
progress = market.get('progress', 1)
if progress and progress < 0.8:
severity = 'high' if progress < 0.6 else 'medium'
insights.append({
'id': f'market_{market.get("market", "unknown")}_underperforming',
'category': 'market',
'severity': severity,
'title': f'{market.get("market", "A")} market underperforming',
'narrative': (
f'{market.get("market", "A")} market is at {round(progress * 100)}% of target. '
f'Focus on lead response time and proposal turnaround.'
),
'action_url': '/admin/multi-market',
})
# Rule 2: At-risk deals in forecast
if forecast_data and isinstance(forecast_data, dict):
at_risk_count = forecast_data.get('at_risk_deals', 0) or 0
at_risk_value = forecast_data.get('at_risk_value', 0) or 0
if at_risk_value > 50000:
insights.append({
'id': 'forecast_at_risk_deals',
'category': 'forecast',
'severity': 'high' if at_risk_value > 100000 else 'medium',
'title': 'At-risk deals in pipeline',
'narrative': (
f'${at_risk_value:,.0f} at risk across {at_risk_count} deals. '
f'Prioritize follow-ups on high-value opportunities.'
),
'action_url': '/admin/forecasting',
})
# Rule 3: Pipeline health score < 70
health_score = forecast_data.get('health_score', 100)
if health_score and health_score < 70:
insights.append({
'id': 'forecast_health_score_low',
'category': 'forecast',
'severity': 'high' if health_score < 50 else 'medium',
'title': 'Pipeline health is low',
'narrative': (
f'Pipeline health score is {health_score}. '
f'Focus on moving deals through the pipeline and removing blockers.'
),
'action_url': '/admin/forecasting',
})
# Rule 4: Goal progress lag (> 20% behind)
if goals_data and isinstance(goals_data, dict):
goals = goals_data.get('goals', [])
if isinstance(goals, list):
for goal in goals:
progress = goal.get('progress', 1)
if progress and progress < 0.8:
insights.append({
'id': f'goal_{goal.get("id", "unknown")}_lagging',
'category': 'goal',
'severity': 'medium',
'title': f'{goal.get("name", "A")} goal lagging',
'narrative': (
f'{goal.get("name", "A")} goal is at {round(progress * 100)}%. '
f'Focus on activities that drive this metric forward.'
),
'action_url': '/admin/goals',
})
# Rule 5: Scale issues (ROAS < 2.0)
if scale_data and isinstance(scale_data, dict):
roas = scale_data.get('roas', 999)
if roas and roas < 2.0:
insights.append({
'id': 'scale_roas_low',
'category': 'scale',
'severity': 'medium',
'title': 'Return on ad spend is low',
'narrative': (
f'ROAS is ${roas:.1f} (target: $2.00). '
f'Review ad campaigns and optimize targeting.'
),
'action_url': '/admin/scale-optimization',
})
# Rule 6: Variance alerts
if intelligence_data and isinstance(intelligence_data, dict):
variance = intelligence_data.get('variance_pct', 0)
if variance and variance < -15:
insights.append({
'id': 'intelligence_variance_negative',
'category': 'intelligence',
'severity': 'high' if variance < -25 else 'medium',
'title': 'Revenue variance below target',
'narrative': (
f'Revenue is {round(variance)}% below forecast. '
f'Review deal progression and identify at-risk opportunities.'
),
'action_url': '/admin/strategic-intelligence',
})
# Sort by severity (high > medium > low)
severity_order = {'high': 0, 'medium': 1, 'low': 2}
insights.sort(key=lambda x: severity_order.get(x.get('severity', 'low'), 3))
return insights
# ── Public demo insight (for landing page) ──────────────────────────
def _demo_insight() -> dict:
"""Return a demo insight for public/landing page access."""
return {
'id': 'demo_insight',
'category': 'market',
'severity': 'medium',
'title': 'Coach the Phoenix market on follow-up speed',
'narrative': 'Recover an estimated $84K this quarter',
'action_url': '#',
'dismissed': False,
}
# ── Routes ───────────────────────────────────────────────────────────
@coaching_bp.route('/insights')
def get_insights():
"""Get coaching insights for the organization.
Public access returns a demo insight for the landing page.
Authenticated access returns real insights based on organizational data.
"""
from flask_login import current_user
# Check cache first
cache_key = 'coaching_insights'
if not current_user.is_authenticated:
# Landing page - return demo insight
return jsonify({
'insights': [_demo_insight()],
'overall_score': 72,
})
cached = _cache_get(cache_key)
if cached:
return jsonify({**cached, 'cache_age_seconds': 0})
# Authenticated - get real insights
token = request.headers.get('Authorization', '').replace('Bearer ', '')
# Evaluate rules
insights = evaluate_rules(token)
# Mark as not dismissed
for insight in insights:
insight['dismissed'] = False
# Calculate overall health score
if insights:
high_count = sum(1 for i in insights if i['severity'] == 'high')
medium_count = sum(1 for i in insights if i['severity'] == 'medium')
overall_score = max(30, 100 - (high_count * 20) - (medium_count * 10))
else:
overall_score = 100
result = {
'insights': insights,
'overall_score': overall_score,
}
# Cache the result
_cache_set(cache_key, result)
return jsonify(result)
@coaching_bp.route('/insights/<insight_id>/dismiss', methods=['POST'])
@require_auth_json()
@require_csrf
def dismiss_insight(insight_id: str):
"""Dismiss a coaching insight (prevents it from showing for 24h)."""
# TODO: Store dismissed insights in DB with timestamp
# For now, just return success
return jsonify({'success': True, 'dismissed': insight_id})
@coaching_bp.route('/score')
def get_score():
"""Get overall coaching score for the organization."""
from flask_login import current_user
if not current_user.is_authenticated:
return jsonify({'overall_score': 72})
cache_key = 'coaching_insights'
cached = _cache_get(cache_key)
if cached:
return jsonify({'overall_score': cached['overall_score']})
token = request.headers.get('Authorization', '').replace('Bearer ', '')
insights = evaluate_rules(token)
if insights:
high_count = sum(1 for i in insights if i['severity'] == 'high')
medium_count = sum(1 for i in insights if i['severity'] == 'medium')
overall_score = max(30, 100 - (high_count * 20) - (medium_count * 10))
else:
overall_score = 100
return jsonify({'overall_score': overall_score})