scripts/extract_trades.py

#!/usr/bin/env python3
"""
Extract Timothy Sykes trades from profit.ly
Two-phase approach:
  Phase 1: Collect all page URLs
  Phase 2: Extract trades from each page
Saves to CSV with resume capability
"""

import argparse
import csv
import logging
import os
import random
import re
import time
from dataclasses import dataclass

import requests
from bs4 import BeautifulSoup

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

# Constants
BASE_URL = "https://profit.ly/user/timothysykes/trades"
CSV_FILE = "data/trades.csv"
PAGE_SIZE = 25
MAX_RETRIES = 3
INITIAL_RETRY_DELAY = 5.0

# Rate limiting (randomized between min/max)
MIN_DELAY = 2.0
MAX_DELAY = 5.0

CSV_FIELDS = ["profit", "ticker", "date", "username", "entry_comments", "exit_comments"]

HEADERS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
]


@dataclass
class Trade:
    """Represents a single trade."""

    profit: str
    ticker: str
    date: str
    username: str
    entry_comments: str
    exit_comments: str

    def to_dict(self) -> dict[str, str]:
        return {
            "profit": self.profit,
            "ticker": self.ticker,
            "date": self.date,
            "username": self.username,
            "entry_comments": self.entry_comments,
            "exit_comments": self.exit_comments,
        }


def get_total_pages(html: str) -> int | None:
    """Extract total pages from pagination control."""
    soup = BeautifulSoup(html, "html.parser")
    input_elem = soup.find("input", {"id": "paginationInput"})
    if input_elem:
        max_val = input_elem.get("max")
        if max_val is not None:
            try:
                return int(str(max_val))
            except (ValueError, TypeError):
                pass
    return None


def extract_trades(html: str) -> list[Trade]:
    """Extract all trades from HTML content using BeautifulSoup."""
    trades = []
    soup = BeautifulSoup(html, "html.parser")

    # Find all trade feed bodies
    feed_bodies = soup.find_all("section", class_="trade-feed-body")

    for feed in feed_bodies:
        # Extract profit from next card-header (it comes AFTER the feed body in HTML)
        card_header = feed.find_next("div", class_="card-header")
        if not card_header:
            continue

        profit_elem = card_header.find("a", class_=["trade-up", "trade-down"])
        ticker_elem = card_header.find("a", class_="trade-ticker")

        if not profit_elem or not ticker_elem:
            continue

        # Extract ticker and type
        ticker = ticker_elem.get_text(strip=True)
        profit = profit_elem.get_text(strip=True)

        # Extract date
        date_elem = feed.find("date")
        date = date_elem.get_text(strip=True) if date_elem else ""

        # Extract username
        username_elem = feed.find("a", class_="name")
        username = username_elem.get_text(strip=True) if username_elem else ""

        # Extract comments
        wall_texts = feed.find_all("p", class_="wall-text")
        entry = ""
        exit_comments = ""

        if wall_texts:
            entry_text = wall_texts[0].get_text(strip=True)
            entry = re.sub(r"^Entry comments:\s*", "", entry_text).strip()

        if len(wall_texts) >= 2:
            exit_text = wall_texts[1].get_text(strip=True)
            exit_comments = re.sub(r"^Exit comments:\s*", "", exit_text).strip()

        trades.append(
            Trade(
                profit=profit,
                ticker=ticker,
                date=date,
                username=username,
                entry_comments=entry,
                exit_comments=exit_comments,
            )
        )

    return trades


def write_csv_header() -> None:
    """Write CSV header if file doesn't exist."""
    os.makedirs(os.path.dirname(CSV_FILE), exist_ok=True)
    if not os.path.exists(CSV_FILE):
        with open(CSV_FILE, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
            writer.writeheader()


# Global set to track seen trades (profit+ticker+date)
_seen_trades: set[tuple[str, str, str]] = set()


def load_existing_trades() -> None:
    """Load existing trades from CSV into the seen set to avoid duplicates."""
    if os.path.exists(CSV_FILE):
        with open(CSV_FILE, newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                _seen_trades.add((row["profit"], row["ticker"], row["date"]))


def append_trades(trades: list[Trade]) -> int:
    """Append trades to CSV file, skipping duplicates. Returns count appended."""
    count = 0
    with open(CSV_FILE, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
        for trade in trades:
            key = (trade.profit, trade.ticker, trade.date)
            if key not in _seen_trades:
                _seen_trades.add(key)
                writer.writerow(trade.to_dict())
                count += 1
    return count


def fetch_page(session: requests.Session, page: int) -> str:
    """Fetch a single page with retry logic."""
    url = f"{BASE_URL}?page={page}&size={PAGE_SIZE}"
    ua = random.choice(HEADERS)
    session.headers.update({"User-Agent": ua})

    for attempt in range(MAX_RETRIES):
        try:
            response = session.get(url, timeout=30)
            response.raise_for_status()
            return response.text

        except requests.HTTPError as e:
            if e.response.status_code == 429:
                delay = min(INITIAL_RETRY_DELAY * (2**attempt), 60)
                logger.warning(f"Rate limited (429), waiting {delay}s...")
                time.sleep(delay)
            elif e.response.status_code >= 500:
                delay = INITIAL_RETRY_DELAY * (2**attempt)
                logger.warning(f"Server error {e.response.status_code}, retry {attempt + 1}/{MAX_RETRIES}")
                time.sleep(delay)
            else:
                raise

        except requests.RequestException as e:
            delay = INITIAL_RETRY_DELAY * (2**attempt)
            logger.warning(f"Request failed, retry {attempt + 1}/{MAX_RETRIES}: {e}")
            time.sleep(delay)

    raise RuntimeError(f"Failed to fetch page {page} after {MAX_RETRIES} attempts")


def phase1_collect_urls(output_dir: str = "data", max_pages: int | None = None) -> None:
    """Phase 1: Download raw HTML files for all pages."""
    session = requests.Session()

    # Fetch first page to get total pages
    logger.info("Phase 1: Discovering total pages...")
    html = fetch_page(session, 1)
    total_pages = get_total_pages(html)

    if not total_pages:
        logger.error("Could not determine total pages, using fallback detection")
        total_pages = 0

    # Limit pages if specified (for testing)
    if max_pages is not None:
        total_pages = min(total_pages, max_pages)

    logger.info(f"Total pages to fetch: {total_pages}")

    # Create output directory
    os.makedirs(output_dir, exist_ok=True)

    # Save first page HTML
    with open(os.path.join(output_dir, "page_1.html"), "w", encoding="utf-8") as f:
        f.write(html)
    logger.info("Saved page 1")

    # Fetch remaining pages
    for page in range(2, total_pages + 1):
        try:
            html = fetch_page(session, page)
            output_file = os.path.join(output_dir, f"page_{page}.html")
            with open(output_file, "w", encoding="utf-8") as f:
                f.write(html)
            logger.info(f"Saved page {page}/{total_pages}")

            # Rate limiting
            delay = random.uniform(MIN_DELAY, MAX_DELAY)
            time.sleep(delay)

        except KeyboardInterrupt:
            logger.info(f"Interrupted at page {page}")
            break
        except Exception as e:
            logger.error(f"Error on page {page}: {e}")
            time.sleep(10)

    logger.info(f"Phase 1 complete: {total_pages} HTML files saved to {output_dir}")


def phase2_extract_trades(input_dir: str = "data") -> None:
    """Phase 2: Read saved HTML files and extract trades to CSV."""
    # Initialize CSV
    write_csv_header()
    load_existing_trades()

    # Find all HTML files
    html_files = sorted(
        [f for f in os.listdir(input_dir) if f.startswith("page_") and f.endswith(".html")],
        key=lambda x: int(x.split("_")[1].split(".")[0]),
    )

    if not html_files:
        logger.error(f"No HTML files found in {input_dir}. Run phase 1 first.")
        return

    logger.info(f"Phase 2: Processing {len(html_files)} HTML files from {input_dir}")

    processed_count = 0
    for html_file in html_files:
        try:
            file_path = os.path.join(input_dir, html_file)
            with open(file_path, encoding="utf-8") as f:
                html = f.read()

            trades = extract_trades(html)
            if trades:
                page_num = html_file.replace("page_", "").replace(".html", "")
                count = append_trades(trades)
                logger.info(f"Page {page_num}: {count} new trades")

            processed_count += 1

        except KeyboardInterrupt:
            logger.info(f"Interrupted at {html_file}")
            break
        except Exception as e:
            logger.error(f"Error on {html_file}: {e}")

    logger.info(f"Phase 2 complete: {processed_count} pages processed")


def main() -> None:
    parser = argparse.ArgumentParser(description="Extract Timothy Sykes trades")
    parser.add_argument(
        "phase",
        choices=["1", "2", "both"],
        default="both",
        nargs="?",
        help="Phase to run: 1 (collect URLs), 2 (extract trades), both (default)",
    )
    parser.add_argument(
        "--max-pages",
        type=int,
        default=None,
        help="Maximum pages to fetch in phase 1 (for testing)",
    )
    args = parser.parse_args()

    if args.phase in ("1", "both"):
        phase1_collect_urls(max_pages=args.max_pages)
    if args.phase in ("2", "both"):
        phase2_extract_trades()


if __name__ == "__main__":
    main()