#!/usr/bin/env python3 import eden_brain_inject """ EDEN MARKET INTELLIGENCE SYSTEM Real-time monitoring of: - Tech news & trends - Startup funding (Crunchbase-style) - VC activity & money flow - Who's buying what - Market sentiment """ import json import time import requests from datetime import datetime, timedelta from pathlib import Path DATA_DIR = "/Eden/DATA" LOG_PATH = f"{DATA_DIR}/market_intelligence.log" STATE_PATH = f"{DATA_DIR}/market_intelligence.json" class MarketIntelligence: def __init__(self): self.state = self.load_state() def load_tech_feeds(self) -> dict: """Load tech feeds data""" try: with open("/Eden/DATA/tech_feeds.json") as f: return json.load(f) except: return {} def load_state(self): try: with open(STATE_PATH) as f: return json.load(f) except: return { "last_update": None, "news": [], "funding_rounds": [], "trending_tech": [], "sentiment": "neutral", "hot_sectors": [] } def save_state(self): self.state['last_update'] = datetime.now().isoformat() with open(STATE_PATH, 'w') as f: json.dump(self.state, f, indent=2, default=str) def log(self, msg): timestamp = datetime.now().strftime("%H:%M:%S") line = f"[{timestamp}] {msg}" print(line) with open(LOG_PATH, 'a') as f: f.write(line + "\n") def fetch_tech_news(self) -> list: """Fetch tech news from HackerNews API (free, no key needed)""" news = [] try: # Get top stories from HackerNews resp = requests.get( "https://hacker-news.firebaseio.com/v0/topstories.json", timeout=10 ) story_ids = resp.json()[:30] # Top 30 for sid in story_ids[:15]: # Process top 15 try: story = requests.get( f"https://hacker-news.firebaseio.com/v0/item/{sid}.json", timeout=5 ).json() if story and story.get('title'): news.append({ 'title': story.get('title', ''), 'url': story.get('url', ''), 'score': story.get('score', 0), 'source': 'hackernews', 'time': datetime.fromtimestamp(story.get('time', 0)).isoformat() }) except: pass self.log(f"📰 Fetched {len(news)} HackerNews stories") except Exception as e: self.log(f"❌ HackerNews error: {e}") return news def fetch_github_trending(self) -> list: """Check GitHub trending repos (indicates hot tech)""" trending = [] try: # GitHub search for recently created popular repos resp = requests.get( "https://api.github.com/search/repositories", params={ 'q': 'created:>2025-11-01 stars:>100', 'sort': 'stars', 'order': 'desc', 'per_page': 20 }, headers={'Accept': 'application/vnd.github.v3+json'}, timeout=15 ) if resp.status_code == 200: for repo in resp.json().get('items', [])[:10]: trending.append({ 'name': repo.get('full_name', 'unknown'), 'description': (repo.get('description') or '')[:100], 'stars': repo['stargazers_count'], 'language': repo.get('language', 'Unknown'), 'url': repo['html_url'], 'topics': repo.get('topics', [])[:5] }) self.log(f"🔥 Found {len(trending)} trending repos") except Exception as e: self.log(f"❌ GitHub trending error: {e}") return trending def analyze_funding_signals(self) -> list: """Look for funding signals in news""" funding_keywords = [ 'raises', 'funding', 'series a', 'series b', 'series c', 'venture', 'million', 'billion', 'investment', 'acquired', 'ipo', 'valuation', 'unicorn' ] signals = [] for news_item in self.state.get('news', []): title_lower = news_item.get('title', '').lower() for keyword in funding_keywords: if keyword in title_lower: signals.append({ 'title': news_item['title'], 'signal': keyword, 'url': news_item.get('url', ''), 'score': news_item.get('score', 0) }) break self.log(f"💰 Found {len(signals)} funding signals") return signals def identify_hot_sectors(self) -> list: """Identify trending tech sectors""" sector_keywords = { 'AI/ML': ['ai', 'ml', 'machine learning', 'llm', 'gpt', 'neural', 'deep learning', 'transformer'], 'Crypto/Web3': ['crypto', 'blockchain', 'web3', 'defi', 'nft', 'bitcoin', 'ethereum'], 'Cloud/Infra': ['cloud', 'aws', 'kubernetes', 'docker', 'serverless', 'devops'], 'Security': ['security', 'cybersecurity', 'hack', 'vulnerability', 'zero-day'], 'Fintech': ['fintech', 'payments', 'banking', 'stripe', 'trading'], 'Dev Tools': ['developer', 'ide', 'api', 'sdk', 'framework', 'library'], 'Data': ['data', 'analytics', 'database', 'sql', 'warehouse'], 'Mobile': ['ios', 'android', 'mobile', 'app', 'react native', 'flutter'] } sector_scores = {sector: 0 for sector in sector_keywords} # Score based on news for news_item in self.state.get('news', []): title_lower = news_item.get('title', '').lower() for sector, keywords in sector_keywords.items(): for kw in keywords: if kw in title_lower: sector_scores[sector] += news_item.get('score', 1) # Score based on trending repos for repo in self.state.get('trending_tech', []): desc = ((repo.get('description') or '') + ' ' + (repo.get('language') or '')).lower() topics = ' '.join(repo.get('topics', [])) combined = desc + ' ' + topics for sector, keywords in sector_keywords.items(): for kw in keywords: if kw in combined: sector_scores[sector] += repo.get('stars', 0) / 100 # Sort by score hot = sorted( [(sector, int(score)) for sector, score in sector_scores.items() if score > 0], key=lambda x: x[1], reverse=True ) self.log(f"🔥 Hot sectors: {[h[0] for h in hot[:3]]}") return hot[:5] def generate_leads_from_intel(self) -> list: """Generate business leads from market intelligence""" leads = [] # Companies mentioned in funding news = potential customers for signal in self.state.get('funding_rounds', []): leads.append({ 'source': 'funding_news', 'reason': f"Recently raised funding: {signal.get('signal', '')}", 'title': signal.get('title', ''), 'url': signal.get('url', ''), 'priority': 'high' # Funded companies have budget! }) # Trending repos = active developers who might need help for repo in self.state.get('trending_tech', [])[:5]: leads.append({ 'source': 'github_trending', 'reason': f"Hot project with {repo.get('stars', 0)} stars", 'title': repo.get('name', ''), 'url': repo.get('url', ''), 'priority': 'medium' }) self.log(f"🎯 Generated {len(leads)} leads from market intel") return leads def calculate_sentiment(self) -> str: """Calculate overall market sentiment""" positive_words = ['raises', 'growth', 'launch', 'success', 'profit', 'breakthrough', 'innovation'] negative_words = ['layoff', 'crash', 'fail', 'loss', 'decline', 'shutdown', 'bankrupt'] pos_count = 0 neg_count = 0 for news_item in self.state.get('news', []): title_lower = news_item.get('title', '').lower() for word in positive_words: if word in title_lower: pos_count += 1 for word in negative_words: if word in title_lower: neg_count += 1 if pos_count > neg_count * 1.5: return "bullish" elif neg_count > pos_count * 1.5: return "bearish" else: return "neutral" def run_cycle(self): """Run full intelligence gathering cycle""" self.log("=" * 50) self.log("🧠 MARKET INTELLIGENCE CYCLE") self.log("=" * 50) # Gather data # Combine HackerNews with tech feeds self.state['news'] = self.fetch_tech_news() # Merge in tech feeds tech_feeds = self.load_tech_feeds() for source in ['reddit', 'lobsters', 'devto', 'techcrunch', 'ycombinator']: for item in tech_feeds.get(source, [])[:10]: self.state['news'].append({ 'title': item.get('title', ''), 'url': item.get('url', ''), 'score': item.get('score', item.get('points', item.get('reactions', 50))), 'source': item.get('source', source) }) time.sleep(2) # Rate limiting self.state['trending_tech'] = self.fetch_github_trending() time.sleep(2) # Analyze self.state['funding_rounds'] = self.analyze_funding_signals() self.state['hot_sectors'] = self.identify_hot_sectors() self.state['sentiment'] = self.calculate_sentiment() # Generate leads leads = self.generate_leads_from_intel() # Save state self.save_state() # Summary self.log("") self.log("📊 INTELLIGENCE SUMMARY:") self.log(f" News items: {len(self.state['news'])}") self.log(f" Trending repos: {len(self.state['trending_tech'])}") self.log(f" Funding signals: {len(self.state['funding_rounds'])}") self.log(f" Hot sectors: {[h[0] for h in self.state['hot_sectors'][:3]]}") self.log(f" Market sentiment: {self.state['sentiment'].upper()}") self.log(f" New leads generated: {len(leads)}") return self.state def run_continuous(self, interval_minutes=30): """Run continuously""" self.log("🧠 Market Intelligence System Started") self.log(f" Update interval: {interval_minutes} minutes") while True: try: self.run_cycle() except Exception as e: self.log(f"❌ Cycle error: {e}") self.log(f"💤 Next update in {interval_minutes} minutes...") time.sleep(interval_minutes * 60) def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument('--once', action='store_true') parser.add_argument('--interval', type=int, default=30) args = parser.parse_args() intel = MarketIntelligence() if args.once: intel.run_cycle() else: intel.run_continuous(args.interval) if __name__ == "__main__": main()