audience-deep-dive

Audience Deep-Dive — LinkedIn Audience Simulator Skill

Purpose: Explore a single segment in detail to understand content preferences and engagement patterns.

Input

{
  "segment_name": "string (e.g., 'Executives', 'Technical Leaders')"
}

Output

{
  "segment": "Executives",
  "segment_size": 150,
  "activity_breakdown": {
    "active": 108,
    "occasional": 27,
    "dormant": 15
  },
  "seniority_distribution": {
    "VP": 64,
    "C-Level": 42,
    "Director": 38,
    "Founder": 6
  },
  "top_roles": ["CEO", "CTO", "CFO", "VP Engineering", "VP Product"],
  "industries": ["Tech", "Finance", "Healthcare"],
  "key_characteristics": [
    "High engagement rate (8.2% vs network avg 2.1%)",
    "Value authenticity and insight",
    "Rarely engage with promotional content",
    "Post frequently about leadership and culture"
  ],
  "topic_engagement": [
    {
      "topic": "Leadership",
      "affinity": 0.92,
      "observed_posts": 847,
      "avg_engagement": 12,
      "why_they_engage": "Stories about building teams, culture, managing hard times"
    },
    {
      "topic": "Strategy",
      "affinity": 0.87,
      "observed_posts": 634,
      "avg_engagement": 11,
      "why_they_engage": "Long-form thoughts on company direction, market trends"
    }
  ],
  "your_past_performance": [
    {
      "post_topic": "Building teams",
      "predicted_reactions": 18,
      "actual_reactions": 22,
      "error": 4,
      "hook_strength": 78,
      "status": "better_than_expected"
    }
  ],
  "content_recommendations": {
    "high_performing_hooks": ["story", "personal", "question"],
    "do_posts": [
      "Write about leadership, team culture, fundraising",
      "Use story or question hooks",
      "Lead with vulnerability or contrarian take"
    ],
    "dont_posts": [
      "Avoid tutorials and how-tos",
      "Avoid heavy emoji or casual tone",
      "Avoid hard sales pitches"
    ]
  },
  "dormant_members": [
    {
      "name": "@Sarah_CEO",
      "reason": "No engagement in 8 months, but historically high engagement",
      "last_engagement": "Leadership content about building teams",
      "reactivation_suggestion": "Post about team building or fundraising journey"
    }
  ],
  "segment_recommendations": [
    "This segment is your most engaged. Focus on leadership and strategy content.",
    "Consider creating a monthly 'founder/CEO insights' series."
  ]
}

Implementation (Python)

def get_deep_dive(segment_name):
    """Main skill function."""
    conn = get_db_connection()

    try:
        cursor = conn.cursor()

        # Get segment
        cursor.execute('SELECT id, name, estimated_size FROM audience_segments WHERE name = ?', (segment_name,))
        seg = cursor.fetchone()

        if not seg:
            return {'error': f'Segment "{segment_name}" not found'}

        segment_id = seg['id']
        segment_size = seg['estimated_size']

        # Activity breakdown
        cursor.execute('''
            SELECT activity_status, COUNT(*) as count
            FROM connection_profiles
            WHERE segment_id = ?
            GROUP BY activity_status
        ''', (segment_id,))

        activity = {row['activity_status']: row['count'] for row in cursor.fetchall()}

        # Seniority distribution
        cursor.execute('''
            SELECT parsed_seniority, COUNT(*) as count
            FROM connection_profiles
            WHERE segment_id = ?
            GROUP BY parsed_seniority
            ORDER BY count DESC
        ''', (segment_id,))

        seniority = {row['parsed_seniority']: row['count'] for row in cursor.fetchall()}

        # Top roles
        cursor.execute('''
            SELECT parsed_role, COUNT(*) as count
            FROM connection_profiles
            WHERE segment_id = ?
            GROUP BY parsed_role
            ORDER BY count DESC
            LIMIT 5
        ''', (segment_id,))

        top_roles = [row['parsed_role'] for row in cursor.fetchall()]

        # Industries
        cursor.execute('''
            SELECT DISTINCT parsed_industry
            FROM connection_profiles
            WHERE segment_id = ?
            LIMIT 5
        ''', (segment_id,))

        industries = [row['parsed_industry'] for row in cursor.fetchall()]

        # Topic engagement
        cursor.execute('''
            SELECT topic_name, affinity_score, posts_count, avg_engagement
            FROM segment_topic_affinity
            WHERE segment_id = ?
            ORDER BY affinity_score DESC
            LIMIT 10
        ''', (segment_id,))

        topic_engagement = []
        for row in cursor.fetchall():
            topic_engagement.append({
                'topic': row['topic_name'],
                'affinity': round(row['affinity_score'], 2),
                'observed_posts': row['posts_count'],
                'avg_engagement': round(row['avg_engagement'], 1),
                'why_they_engage': get_topic_engagement_reason(segment_name, row['topic_name'])
            })

        # Key characteristics (heuristic)
        characteristics = [
            f"Activity level: {int(activity.get('active', 0) / segment_size * 100)}% active",
            f"Top seniority: {list(seniority.keys())[0] if seniority else 'Unknown'}",
            f"Primary industries: {', '.join(industries) if industries else 'Diverse'}",
            f"Strongest topic affinity: {topic_engagement[0]['topic'] if topic_engagement else 'Unknown'} ({topic_engagement[0]['affinity']*100:.0f}%)" if topic_engagement else "No affinity data"
        ]

        # Past performance with this segment
        cursor.execute('''
            SELECT draft_text, overall_score, predicted_reactions, actual_reactions
            FROM predictions
            WHERE json_extract(top_segments_json, '$[0].segment') = ?
            ORDER BY scored_at DESC
            LIMIT 5
        ''', (segment_name,))

        past_performance = []
        for row in cursor.fetchall():
            if row['actual_reactions'] is not None:
                error = row['predicted_reactions'] - row['actual_reactions']
                past_performance.append({
                    'topic': row['draft_text'][:50] + '...',
                    'predicted_reactions': row['predicted_reactions'],
                    'actual_reactions': row['actual_reactions'],
                    'error': error,
                    'status': 'better_than_expected' if error < 0 else 'worse_than_expected'
                })

        # Content recommendations
        recommendations = {
            'high_performing_hooks': get_high_performing_hooks(cursor, segment_name),
            'do_posts': get_do_recommendations(segment_name),
            'dont_posts': get_dont_recommendations(segment_name)
        }

        # Dormant members
        cursor.execute('''
            SELECT name, headline, engagement_score
            FROM connection_profiles
            WHERE segment_id = ? AND activity_status = 'dormant'
            ORDER BY engagement_score DESC
            LIMIT 5
        ''', (segment_id,))

        dormant_members = []
        for row in cursor.fetchall():
            dormant_members.append({
                'name': row['name'],
                'reason': f"No recent engagement, but historically active ({row['engagement_score']:.1f} avg engagement)",
                'reactivation_suggestion': get_reactivation_suggestion(segment_name, row['headline'])
            })

        return {
            'segment': segment_name,
            'segment_size': segment_size,
            'activity_breakdown': activity,
            'seniority_distribution': seniority,
            'top_roles': top_roles,
            'industries': industries,
            'key_characteristics': characteristics,
            'topic_engagement': topic_engagement,
            'your_past_performance': past_performance,
            'content_recommendations': recommendations,
            'dormant_members': dormant_members,
            'segment_recommendations': [
                f"This is a {get_segment_assessment(activity, segment_size)} segment. Focus on {topic_engagement[0]['topic'] if topic_engagement else 'relevant'} content.",
                "Run /score to see how a draft performs with this segment specifically."
            ]
        }

    except Exception as e:
        return {'error': f'Error: {str(e)}'}

    finally:
        conn.close()

def get_topic_engagement_reason(segment, topic):
    """Return why a segment engages with a topic."""
    reasons = {
        ('Executives', 'Leadership'): 'Stories about building teams, culture, managing in hard times',
        ('Executives', 'Strategy'): 'Long-form thoughts on company direction, market trends',
        ('Technical Leaders', 'AI/ML'): 'Deep dives into models, algorithms, LLMs, RAG, transformers',
        ('Technical Leaders', 'Cloud'): 'Infrastructure, Kubernetes, serverless, DevOps patterns',
        ('Growth & Sales', 'Growth Hacking'): 'Metrics, experiments, scaling, acquisition strategies',
        ('Growth & Sales', 'Revenue'): 'Pricing, expansion, upsell, revenue models',
    }
    return reasons.get((segment, topic), f'Content about {topic}')

def get_high_performing_hooks(cursor, segment_name):
    """Get hooks that work best for this segment."""
    hook_map = {
        'Executives': ['story', 'personal', 'contrarian'],
        'Technical Leaders': ['data-driven', 'question', 'listicle'],
        'Growth & Sales': ['data-driven', 'story', 'listicle'],
        'General Audience': ['story', 'question', 'personal']
    }
    return hook_map.get(segment_name, ['story', 'question'])

def get_do_recommendations(segment_name):
    """Get content recommendations for a segment."""
    do_map = {
        'Executives': [
            "Write about leadership, team culture, fundraising, strategy",
            "Use story or question hooks",
            "Lead with vulnerability or contrarian take",
            "Include real data, but always with business context"
        ],
        'Technical Leaders': [
            "Deep-dive into technical topics: AI/ML, Cloud, DevOps, Product",
            "Lead with data or technical question",
            "Include code examples, architecture diagrams, benchmarks",
            "Challenge best practices with data-driven arguments"
        ],
        'Growth & Sales': [
            "Write about growth, GTM, revenue, scaling strategies",
            "Lead with metrics or story",
            "Include case studies, benchmarks, experimentation results",
            "Focus on what worked and how to replicate it"
        ]
    }
    return do_map.get(segment_name, ["Post authentic, relevant content"])

def get_dont_recommendations(segment_name):
    """Get what NOT to post for a segment."""
    dont_map = {
        'Executives': [
            "Avoid tutorials and how-tos (too junior-level)",
            "Avoid heavy emoji or overly casual tone",
            "Avoid hard sales pitches and promotional content",
            "Avoid pure technical deep-dives without business context"
        ],
        'Technical Leaders': [
            "Avoid non-technical content (they skim quickly)",
            "Avoid vague statements without data or examples",
            "Avoid hard sells and promotional spam",
            "Avoid leadership/business posts that ignore technical reality"
        ],
        'Growth & Sales': [
            "Avoid pure technical content (too specialized)",
            "Avoid theoretical discussions without business outcomes",
            "Avoid leadership-focused content without growth angle",
            "Avoid unsubstantiated claims"
        ]
    }
    return dont_map.get(segment_name, ["Avoid low-quality or off-brand content"])

def get_reactivation_suggestion(segment_name, headline):
    """Get a suggestion to reactivate this dormant member."""
    if 'engineer' in headline.lower():
        return "Post about AI/ML or Cloud topics they likely care about"
    elif 'executive' in headline.lower() or 'ceo' in headline.lower():
        return "Share a founder story or leadership insight"
    else:
        return "Post about topics this person engaged with before"

def get_segment_assessment(activity, total_size):
    """Assess segment quality."""
    active_pct = activity.get('active', 0) / total_size if total_size > 0 else 0
    if active_pct > 0.7:
        return "highly engaged"
    elif active_pct > 0.5:
        return "moderately engaged"
    else:
        return "low engagement"