DilutionTracker APIdilutiontracker.com ↗

"""
DilutionTracker API Client
Get your API key from: https://parse.bot/settings

A practical example showing how to use the DilutionTracker API to search
for stocks, analyze dilution events, and track market trends.
"""

import os
import requests
from typing import Any, Dict, Optional, List


class ParseClient:
    """Client for interacting with the DilutionTracker API via Parse."""
    
    def __init__(self, api_key: Optional[str] = None):
        """Initialize the ParseClient with API credentials."""
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "ec3cf538-1ae1-4340-b68a-f11c48041584"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError("API key must be provided or set in PARSE_API_KEY environment variable")
    
    def _call(self, endpoint: str, method: str = "POST", **params) -> Dict[str, Any]:
        """Make an API call to the Parse endpoint."""
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        try:
            if method.upper() == "GET":
                response = requests.get(url, headers=headers, params=params, timeout=30)
            else:  # POST
                response = requests.post(url, headers=headers, json=params, timeout=30)
            
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"API call failed: {e}")
            raise
    
    def search_tickers(self, query: str, limit: int = 20) -> List[Dict[str, Any]]:
        """
        Search for tickers by symbol or company name.
        
        Args:
            query: Search keyword (symbol or name)
            limit: Max results to return (default: 20)
        
        Returns:
            List of matching tickers with symbol, companyName, and coverageStatus
        """
        result = self._call("search_tickers", method="GET", query=query, limit=limit)
        if isinstance(result, dict) and "data" in result:
            return result.get("data", [])
        return result if isinstance(result, list) else []
    
    def get_popular_tickers(self) -> List[Dict[str, Any]]:
        """
        Fetch popular tickers currently trending on the platform.
        
        Returns:
            List of popular tickers with market data and performance
        """
        result = self._call("get_popular_tickers", method="GET")
        if isinstance(result, dict) and "data" in result:
            return result.get("data", [])
        return result if isinstance(result, list) else []
    
    def get_open_access_tickers(self) -> List[Dict[str, Any]]:
        """
        Fetch open access tickers available without a subscription.
        
        Returns:
            List of open access tickers with symbol, company name, and CIK
        """
        result = self._call("get_open_access_tickers", method="GET")
        if isinstance(result, dict) and "data" in result:
            return result.get("data", [])
        return result if isinstance(result, list) else []
    
    def get_pending_s1_offerings(self) -> List[Dict[str, Any]]:
        """
        Fetch pending S-1 offerings.
        
        Returns:
            List of companies with pending S-1 offerings
        """
        result = self._call("get_pending_s1_offerings", method="GET")
        if isinstance(result, dict) and "offerings" in result:
            return result.get("offerings", [])
        return result if isinstance(result, list) else []
    
    def get_reverse_splits(self) -> List[Dict[str, Any]]:
        """
        Fetch upcoming and completed reverse splits.
        
        Returns:
            List of reverse split events with symbol and effective date
        """
        result = self._call("get_reverse_splits", method="GET")
        if isinstance(result, dict) and "splits" in result:
            return result.get("splits", [])
        return result if isinstance(result, list) else []


def print_header(text: str) -> None:
    """Print a formatted section header."""
    print(f"\n{'=' * 80}")
    print(f"  {text}")
    print(f"{'=' * 80}")


def main():
    """Demonstrate practical usage of the DilutionTracker API."""
    
    # Initialize client
    try:
        client = ParseClient()
    except ValueError as e:
        print(f"Error: {e}")
        return
    
    # Step 1: Search for a specific ticker to understand coverage
    print_header("Step 1: Search for Specific Companies")
    
    search_queries = ["GameStop", "Tesla", "biotech"]
    all_search_results = {}
    
    for query in search_queries:
        print(f"\n🔍 Searching for '{query}'...")
        try:
            results = client.search_tickers(query, limit=5)
            all_search_results[query] = results
            
            if results:
                print(f"   Found {len(results)} matching ticker(s):")
                for ticker_info in results:
                    symbol = ticker_info.get("symbol", "N/A")
                    company = ticker_info.get("companyName", "N/A")
                    coverage = ticker_info.get("coverageStatus", "N/A")
                    print(f"   • {symbol:8} | {company:35} | Coverage: {coverage}")
            else:
                print(f"   No results found for '{query}'")
        except Exception as e:
            print(f"   Error searching: {e}")
    
    # Step 2: Get popular tickers and identify market trends
    print_header("Step 2: Analyze Popular Tickers and Market Trends")
    
    try:
        popular_tickers = client.get_popular_tickers()
        
        if popular_tickers:
            print(f"\n📈 Found {len(popular_tickers)} trending tickers")
            print("\nTop 10 by activity:")
            print(f"{'#':<3} {'Ticker':<8} {'Company':<35} {'Sector':<20} {'% Change':<10}")
            print("-" * 80)
            
            for i, ticker_info in enumerate(popular_tickers[:10], 1):
                ticker = ticker_info.get("ticker", "N/A")
                company = ticker_info.get("companyName", "N/A")[:32]
                sector = ticker_info.get("sector", "N/A")[:18]
                pct_change = ticker_info.get("pctChangeSinceLastCloseString", "N/A")
                print(f"{i:<3} {ticker:<8} {company:<35} {sector:<20} {pct_change:<10}")
            
            # Analyze sector distribution
            sectors = {}
            for ticker_info in popular_tickers:
                sector = ticker_info.get("sector", "Unknown")
                sectors[sector] = sectors.get(sector, 0) + 1
            
            print("\n📊 Sector Distribution:")
            for sector, count in sorted(sectors.items(), key=lambda x: x[1], reverse=True):
                print(f"   • {sector}: {count} tickers")
        else:
            print("No popular tickers available")
    except Exception as e:
        print(f"Error fetching popular tickers: {e}")
    
    # Step 3: Get open access tickers for free dilution analysis
    print_header("Step 3: Access Free Dilution Data (Open Access Tickers)")
    
    try:
        open_tickers = client.get_open_access_tickers()
        
        if open_tickers:
            print(f"\n🔓 Found {len(open_tickers)} open access tickers (free data)")
            
            # Build industry map
            industries = {}
            cik_map = {}  # Store for later cross-reference
            
            for ticker_info in open_tickers:
                symbol = ticker_info.get("symbol", "")
                industry = ticker_info.get("industry", "Unknown")
                cik = ticker_info.get("cik", "")
                
                industries[industry] = industries.get(industry, 0) + 1
                if symbol:
                    cik_map[symbol] = cik
            
            print("\n📋 Industry Distribution (top 15):")
            print(f"{'#':<3} {'Industry':<40} {'Count':<8}")
            print("-" * 55)
            
            for i, (industry, count) in enumerate(
                sorted(industries.items(), key=lambda x: x[1], reverse=True)[:15], 1
            ):
                print(f"{i:<3} {industry:<40} {count:<8}")
            
            # Show sample tickers
            print("\n📌 Sample Open Access Tickers (first 8):")
            print(f"{'Symbol':<10} {'Company':<40} {'Industry':<25}")
            print("-" * 80)
            
            for ticker_info in open_tickers[:8]:
                symbol = ticker_info.get("symbol", "N/A")
                company = ticker_info.get("companyName", "N/A")[:38]
                industry = ticker_info.get("industry", "N/A")[:23]
                print(f"{symbol:<10} {company:<40} {industry:<25}")
        else:
            print("No open access tickers available")
    except Exception as e:
        print(f"Error fetching open access tickers: {e}")
    
    # Step 4: Monitor dilution events - pending S-1 offerings
    print_header("Step 4: Monitor IPO Activity (Pending S-1 Offerings)")
    
    try:
        s1_offerings = client.get_pending_s1_offerings()
        
        if s1_offerings:
            print(f"\n🚀 Found {len(s1_offerings)} companies with pending S-1 offerings")
            print("\nSample S-1 Offerings (first 10):")
            
            for i, offering in enumerate(s1_offerings[:10], 1):
                symbol = offering.get("symbol", "N/A")
                company = offering.get("companyName", "N/A")
                print(f"   {i:2}. {symbol:8} | {company}")
        else:
            print("No pending S-1 offerings currently tracked")
    except Exception as e:
        print(f"Error fetching S-1 offerings: {e}")
    
    # Step 5: Track reverse splits (dilution risk indicator)
    print_header("Step 5: Track Reverse Splits (Dilution Risk Indicator)")
    
    try:
        reverse_splits = client.get_reverse_splits()
        
        if reverse_splits:
            print(f"\n📉 Found {len(reverse_splits)} reverse split events")
            print("\nRecent Reverse Splits (first 10):")
            
            for i, split in enumerate(reverse_splits[:10], 1):
                symbol = split.get("symbol", "N/A")
                date = split.get("effectiveDate", "N/A")
                ratio = split.get("ratio", "N/A")
                print(f"   {i:2}. {symbol:8} | Date: {date} | Ratio: {ratio}")
        else:
            print("No reverse split data available")
    except Exception as e:
        print(f"Error fetching reverse splits: {e}")
    
    # Step 6: Cross-analysis workflow
    print_header("Step 6: Dilution Risk Analysis - Cross-Reference Check")
    
    try:
        # Get all datasets
        open_access = client.get_open_access_tickers()
        s1_offerings = client.get_pending_s1_offerings()
        reverse_splits = client.get_reverse_splits()
        
        if open_access and (s1_offerings or reverse_splits):
            open_symbols = {t.get("symbol") for t in open_access if t.get("symbol")}
            s1_symbols = {o.get("symbol") for o in s1_offerings if o.get("symbol")}
            split_symbols = {s.get("symbol") for s in reverse_splits if s.get("symbol")}
            
            # Find overlaps
            s1_overlap = open_symbols.intersection(s1_symbols)
            split_overlap = open_symbols.intersection(split_symbols)
            
            print(f"\n📊 Analysis Results:")
            print(f"   Total open access tickers: {len(open_symbols)}")
            print(f"   Total pending S-1 offerings: {len(s1_symbols)}")
            print(f"   Total reverse split events: {len(split_symbols)}")
            
            if s1_overlap:
                print(f"\n⚠️  ALERT: {len(s1_overlap)} open access ticker(s) with pending S-1 offerings:")
                for symbol in sorted(s1_overlap)[:5]:
                    print(f"      • {symbol}")
            else:
                print(f"\n✅ No overlap between open access tickers and pending S-1 offerings")
            
            if split_overlap:
                print(f"\n⚠️  ALERT: {len(split_overlap)} open access ticker(s) with recent reverse splits:")
                for symbol in sorted(split_overlap)[:5]:
                    print(f"      • {symbol}")
            else:
                print(f"\n✅ No open access tickers with recent reverse splits")
            
            # Summary statistics
            print(f"\n📈 Coverage Summary:")
            print(f"   High-risk open access tickers: {len(s1_overlap.union(split_overlap))}")
            print(f"   Safe open access tickers: {len(open_symbols) - len(s1_overlap.union(split_overlap))}")
        else:
            print("Insufficient data for cross-reference analysis")
    except Exception as e:
        print(f"Error during cross-analysis: {e}")
    
    print_header("Analysis Complete")
    print("✅ DilutionTracker API workflow executed successfully!")


if __name__ == "__main__":
    main()