MarketPulse AI

An Event-Driven, Microservice-Based Startup Intelligence Pipeline.

MarketPulse AI is a resilient data pipeline that discovers startups (e.g., via Y Combinator), performs deep-dive scrapes on their actual websites, uses local LLMs to extract structured business intelligence, and stores the data in a Vector Database for natural language semantic search.

---

System Architecture

The platform utilizes a Database-per-Service microservice architecture, orchestrated entirely asynchronously via Apache Kafka to prevent HTTP timeouts during long-running AI and scraping tasks.

graph TD
    subgraph "Clients"
        User[User / API Client]
        Web[Startup Websites]
    end

    subgraph "API Gateway Service"
        Gateway[FastAPI Gateway]
        JobDB[(PostgreSQL\nJob Tracker)]
    end

    subgraph "Message Broker"
        Kafka{Apache Kafka}
    end

    subgraph "Scraper Service"
        Scraper[Selenium Worker]
        MongoDB[(MongoDB\nRaw Data Lake)]
    end

    subgraph "AI Parser Service"
        Parser[LangChain Worker]
        Ollama1((Ollama LLM\nMistral))
    end

    subgraph "Vector Search Service"
        Vector[FastAPI Search]
        Ollama2((Ollama\nEmbeddings))
        PgVector[(PostgreSQL + pgvector)]
    end

    %% Flow of Data
    User -- "1. POST /scrape" --> Gateway
    Gateway -- "2. Track Status" --> JobDB
    Gateway -- "3. Pub: scrape-tasks" --> Kafka

    Kafka -- "4. Sub: scrape-tasks" --> Scraper
    Scraper -- "5. Crawl" --> Web
    Scraper -- "6. Backup HTML" --> MongoDB
    Scraper -- "7. Pub: raw-data" --> Kafka

    Kafka -- "8. Sub: raw-data" --> Parser
    Parser -- "9. Extract JSON" --> Ollama1
    Parser -- "10. Pub: parsed-data" --> Kafka

    Kafka -- "11. Sub: parsed-data" --> Vector
    Vector -- "12. Generate Embeddings" --> Ollama2
    Vector -- "13. Store Intelligence" --> PgVector

    User -- "14. GET /search" --> Vector

Loading

---

Service Roles & Storage

Service	Responsibility	Database	Tech Stack
1. API Gateway	Accepts user scrape requests, tracks job statuses, and publishes tasks.	PostgreSQL (gateway-db)	FastAPI, SQLAlchemy
2. Scraper Service	Multi-stage crawler (Directory -> Profile -> Actual Website).	MongoDB (raw-db)	Selenium, Motor, AsyncIO
3. AI Parser Service	Ingests raw text and extracts structured JSON (Pricing, Audience, Tech).	Stateless	Ollama, LangChain, Pydantic
4. Vector Search Service	Generates embeddings from parsed JSON and serves Semantic Search APIs.	PostgreSQL + pgvector	pgvector, FastAPI, Nomic

---

Project Structure

MarketPulse-AI/
├── docker-compose.yml           # Master orchestrator for services, DBs, and Kafka
├── .env.example                 # Environment variables template
├── api-gateway/                 # Entry point for users (Job creation & status)
├── scraper-service/             # Multi-stage Selenium workers
├── ai-parser-service/           # Local LLM intelligence extraction
└── vector-search-service/       # Embeddings generation & Semantic Search API

---

Quick Start (Docker)

Prerequisites

1. Docker Desktop (or Docker Engine + Compose plugin).
2. Internet access for pulling container images and Python dependencies.
3. Sufficient RAM (16GB+ recommended) for Ollama model inference and Chromium scraping workloads.

Installation

1. Clone the repository and configure your environment variables:

cp .env.example .env

2. Boot up the entire pipeline, including databases and message brokers:

docker compose up -d --build

3. (First Run Only) Download the required AI models into the Ollama container:

docker exec -it marketpulse-ollama ollama pull mistral
docker exec -it marketpulse-ollama ollama pull nomic-embed-text

---

Example Workflow

1. Trigger a Scrape Job

Hit the API Gateway to drop a task into Kafka.

curl -X POST "http://localhost:8000/api/jobs/scrape" \
  -H "Content-Type: application/json" \
  -d '{"batch":"W24", "limit":5}'

2. Monitor Job Status

curl -X GET "http://localhost:8000/api/jobs/<job_id>"

3. Query Semantic Search

Once the data has successfully passed through the Scraper and AI Parser, search the database using natural language via the Vector Service.

curl -X GET "http://localhost:8001/api/search?query=B2B+AI+security+platforms"

Interactive API Docs:

• API Gateway Swagger: http://localhost:8000/docs
• Vector Search Swagger: http://localhost:8001/docs

---

Operational Notes & Troubleshooting

• Postgres Initialization: Logs indicating initdb and fast shutdowns on first boot are expected behavior as Docker provisions the volumes.
• Environment Variables: docker restart does not refresh environment variables. If your .env file changes, recreate the container using:

  docker compose up -d --build --force-recreate <service-name>

• Database Credentials: Vector DB credentials containing special characters (like @ or #) must be URL-encoded (e.g., %40) when used in SQLAlchemy DSNs.
• Headless Scraping: The scraper service utilizes a container-installed Chromium and ChromeDriver for highly stable headless execution. Occasional renderer timeouts on heavy SPA websites are automatically caught and retried.
• Kafka Rebalancing: If a scraping task exceeds 30 minutes, Kafka's max_poll_interval_ms may trigger a consumer rebalance. This limit is intentionally elevated to accommodate slow websites.

---

License

Released under the MIT License.