PIOE / backend /intelligence /silent_detector.py
B1acB1rd
PIOE 2.0 ready for deploymnet
4d92cd5
"""
PIOE Silent Opportunities Detector - Version 2.0
Detects implicit/hidden opportunities that are never announced clearly.
These appear in blog posts, tweets, Discord messages, research updates.
Examples:
- "We're exploring ideas around..."
- "We're looking for collaborators..."
- "If anyone is interested..."
- "We're building something new..."
"""
import re
from typing import Optional
class SilentOpportunityDetector:
"""
Detects implicit opportunities from content that doesn't
explicitly announce them as opportunities.
"""
# Patterns for implicit opportunities
SIGNAL_PATTERNS = {
# Pre-hiring signals
"pre_hiring": [
r"we(?:'re| are) (?:actively )?(?:looking|searching) for",
r"we need (?:a |someone|people)",
r"hiring (?:soon|next|this)",
r"building (?:a |our |the )?team",
r"if you(?:'re| are) interested in joining",
r"open roles? (?:coming|soon)",
r"dm (?:me|us) if (?:you(?:'re| are)|interested)",
r"reach out if",
],
# Pre-grant signals
"pre_grant": [
r"(?:we(?:'re| are)|we will be) (?:funding|supporting|backing)",
r"grants? (?:coming|opening|soon|next)",
r"ecosystem fund",
r"builder(?:s)? program",
r"retroactive (?:funding|rewards)",
r"announcing.{0,30}funding",
r"accepting applications",
],
# Collaboration signals
"collaboration": [
r"looking for (?:collaborators?|partners?|co-founder)",
r"seeking (?:collaborat|partner)",
r"open to (?:collaborat|partner|work)",
r"anyone (?:want|interested).{0,30}(?:build|work|collaborat)",
r"let(?:'s| us) (?:build|work|create) together",
r"who wants to",
r"exploring.{0,30}partnership",
],
# Project/research signals
"research": [
r"we(?:'re| are) (?:exploring|researching|investigating)",
r"new (?:research|project|initiative)",
r"call for (?:papers?|proposals?|abstracts?)",
r"(?:research|academic) (?:collaboration|partnership)",
r"phd (?:position|opportunity|student)",
r"postdoc",
r"looking for (?:interns?|students?)",
],
# Community/ambassador signals
"ambassador": [
r"ambassador program",
r"community (?:lead|manager|role)",
r"help (?:us )?(?:grow|build|spread)",
r"join (?:our|the) (?:community|team|movement)",
r"early (?:adopter|supporter)",
],
# Investment/demo signals
"investment": [
r"demo day",
r"pitch (?:competition|event|day)",
r"investor (?:meeting|demo|call)",
r"raising (?:a |our )?(?:seed|round|series)",
r"open to (?:investment|investors)",
],
}
# Strength indicators (modifiers)
STRENGTH_BOOSTERS = [
r"immediately",
r"urgently",
r"actively",
r"now",
r"today",
r"this week",
r"asap",
r"serious",
r"exciting",
]
# Negative patterns (reduce signal)
NOISE_PATTERNS = [
r"not (?:looking|hiring|seeking)",
r"no longer",
r"was (?:looking|hiring)",
r"used to",
r"back in",
r"years? ago",
r"hypothetically",
r"if only",
]
def detect(self, text: str, title: str = "") -> dict:
"""
Analyze text for silent opportunity signals.
Returns:
- is_silent_opportunity: bool
- opportunity_type: str (pre_hiring, pre_grant, etc.)
- signal_strength: float (0.0 to 1.0)
- detected_patterns: list
- recommended_category: str
"""
full_text = f"{title} {text}".lower()
# Check for noise patterns first
if self._has_noise(full_text):
return {
"is_silent_opportunity": False,
"opportunity_type": None,
"signal_strength": 0.0,
"detected_patterns": [],
"recommended_category": None,
}
# Detect patterns
detected = {}
for opp_type, patterns in self.SIGNAL_PATTERNS.items():
matches = self._find_matches(full_text, patterns)
if matches:
detected[opp_type] = matches
if not detected:
return {
"is_silent_opportunity": False,
"opportunity_type": None,
"signal_strength": 0.0,
"detected_patterns": [],
"recommended_category": None,
}
# Find primary opportunity type
primary_type = max(detected, key=lambda k: len(detected[k]))
# Calculate signal strength
signal_strength = self._calculate_strength(
full_text, detected, primary_type
)
# Map to category
category_map = {
"pre_hiring": "pre_hiring_signal",
"pre_grant": "pre_grant_signal",
"collaboration": "collaboration",
"research": "research",
"ambassador": "ambassador",
"investment": "pitch_event",
}
return {
"is_silent_opportunity": True,
"opportunity_type": primary_type,
"signal_strength": round(signal_strength, 3),
"detected_patterns": detected[primary_type],
"recommended_category": category_map.get(primary_type, "weak_signal"),
}
def _find_matches(self, text: str, patterns: list) -> list:
"""Find all matching patterns in text."""
matches = []
for pattern in patterns:
if re.search(pattern, text, re.IGNORECASE):
# Extract the matching context
match = re.search(pattern, text, re.IGNORECASE)
if match:
# Get surrounding context
start = max(0, match.start() - 20)
end = min(len(text), match.end() + 20)
context = text[start:end]
matches.append(context.strip())
return matches
def _has_noise(self, text: str) -> bool:
"""Check if text contains noise patterns."""
for pattern in self.NOISE_PATTERNS:
if re.search(pattern, text, re.IGNORECASE):
return True
return False
def _calculate_strength(
self,
text: str,
detected: dict,
primary_type: str
) -> float:
"""Calculate signal strength."""
base_strength = 0.5
# More patterns = stronger signal
pattern_count = len(detected[primary_type])
base_strength += min(pattern_count * 0.1, 0.3)
# Check for strength boosters
for booster in self.STRENGTH_BOOSTERS:
if re.search(booster, text, re.IGNORECASE):
base_strength += 0.05
# Multiple types of signals = stronger
if len(detected) > 1:
base_strength += 0.1
# Cap at 1.0
return min(base_strength, 1.0)
def reclassify_opportunity(
self,
opportunity: dict
) -> tuple[str, float]:
"""
Re-evaluate an existing opportunity for silent signals.
Returns (new_category, confidence)
"""
title = opportunity.get("title", "")
text = opportunity.get("raw_text", "")
result = self.detect(text, title)
if result["is_silent_opportunity"]:
return (
result["recommended_category"],
result["signal_strength"]
)
return (None, 0.0)
class OpportunityLanguageDetector:
"""
Detects the urgency, timing, and action language in opportunities.
"""
TIMING_PATTERNS = {
"early": [
r"early (?:bird|access|application)",
r"just (?:launched|announced|opened)",
r"applications? (?:now )?open",
r"first (?:round|batch|cohort)",
r"founding",
r"new program",
],
"optimal": [
r"applications? (?:open|accepted)",
r"deadline (?:is )?(?:soon|approaching)",
r"apply (?:now|today)",
r"last call",
r"extended deadline",
],
"late": [
r"deadline (?:in )?(?:days?|hours?)",
r"closes? (?:soon|tomorrow|today)",
r"final (?:day|hour|chance)",
r"last (?:day|chance)",
],
}
def detect_timing(self, text: str) -> str:
"""Detect application timing."""
text = text.lower()
for timing, patterns in self.TIMING_PATTERNS.items():
for pattern in patterns:
if re.search(pattern, text, re.IGNORECASE):
return timing
return "unknown"
def extract_action_items(self, text: str) -> list:
"""Extract actionable items from text."""
actions = []
# Common action patterns
action_patterns = [
r"apply (?:at|via|through|here)",
r"visit (?:our|the) (?:website|page|link)",
r"(?:fill|submit).{0,20}(?:form|application)",
r"send.{0,20}(?:email|resume|cv|portfolio)",
r"register (?:at|on|here)",
r"sign up",
r"join.{0,20}(?:discord|telegram|slack)",
r"dm (?:me|us)",
r"follow.{0,10}on",
]
for pattern in action_patterns:
match = re.search(pattern, text, re.IGNORECASE)
if match:
start = max(0, match.start() - 10)
end = min(len(text), match.end() + 30)
actions.append(text[start:end].strip())
return actions[:5] # Limit to 5 actions