Method and tooling

#!/usr/bin/env python3
"""
IB Ireland tradability + liquidity gate.

Reads the AIM full list and FTSE SmallCap constituents CSVs, checks each
name against IBKR for a tradable contract (Pass 1: reqContractDetails),
then pulls ~3 months of daily bars for survivors to compute average daily
volume (Pass 2: reqHistoricalData). Writes a single output CSV that feeds
the coverage ledger. This is the hard gate run before any document-fetch
work begins -- it stops at tradability and liquidity, nothing else.

Connection requirement
-----------------------
This script does NOT start TWS or IB Gateway for you. Before running it:
  1. Start TWS or IB Gateway and log in (paper or live).
  2. Configuration -> API -> Settings -> Enable ActiveX and Socket Clients.
  3. Confirm the port matches what you pass with --port:
       paper TWS      7497 (default)
       live TWS       7496
       paper Gateway  4002
       live Gateway   4001

Pacing notes
------------
- The 60-requests-per-10-minutes limit is the one IB documents specifically
  for reqHistoricalData. Pass 2 enforces it with a sliding-window limiter
  (see SlidingWindowLimiter), kept a little under 60 for safety margin.
- reqContractDetails pacing was not independently verified for this script,
  so Pass 1 still throttles conservatively between calls (--contract-throttle)
  and backs off hard on any pacing-violation signal rather than assuming no
  limit applies: error code 100 ("max rate of messages exceeded"), any code
  in the 420-429 range, or any error message containing "pacing".
- reqFundamentalData is never used here. It's deprecated and disqualified as
  a numeric source (registry Source 7) -- this script's job stops at
  tradability and liquidity, not financials.
"""

import argparse
import sys
import time
from collections import deque
from pathlib import Path

import pandas as pd
from ib_async import IB, Contract, Stock

DEFAULT_PORT_HINTS = (
    "paper TWS 7497 | live TWS 7496 | paper Gateway 4002 | live Gateway 4001"
)

# Exchange qualifiers to try in order for a UK small-cap / AIM name.
# SMART+primaryExchange is tried first (usual routing); a direct LSE query
# is the fallback for names SMART won't resolve unambiguously.
EXCHANGE_ATTEMPTS = [
    {"exchange": "SMART", "primaryExchange": "LSE"},
    {"exchange": "LSE", "primaryExchange": ""},
]


class PacingGuard:
    """Watches IB error events for pacing-violation signals and forces a cooldown."""

    def __init__(self, ib):
        self.hit = False
        self.last_code = None
        self.last_msg = ""
        ib.errorEvent += self._on_error

    def _on_error(self, reqId, errorCode, errorString, contract=None):
        is_pacing = (
            errorCode == 100
            or 420 <= errorCode < 430
            or "pacing" in str(errorString).lower()
        )
        if is_pacing:
            self.hit = True
            self.last_code = errorCode
            self.last_msg = errorString

    def check_and_cooldown(self, ib, cooldown_sec):
        if self.hit:
            print(
                f"  [PACING] error {self.last_code}: {self.last_msg} "
                f"-- backing off {cooldown_sec}s"
            )
            ib.sleep(cooldown_sec)
            self.hit = False


class SlidingWindowLimiter:
    """Caps calls to max_calls within a trailing period_sec window."""

    def __init__(self, max_calls, period_sec):
        self.max_calls = max_calls
        self.period_sec = period_sec
        self.calls = deque()

    def wait_if_needed(self, ib):
        now = time.monotonic()
        while self.calls and now - self.calls[0] > self.period_sec:
            self.calls.popleft()
        if len(self.calls) >= self.max_calls:
            sleep_for = self.period_sec - (now - self.calls[0]) + 1
            print(
                f"  [RATE LIMIT] at {self.max_calls}/{self.period_sec}s cap, "
                f"sleeping {sleep_for:.0f}s"
            )
            ib.sleep(sleep_for)
        self.calls.append(time.monotonic())


def load_candidates(aim_path, ftse_path):
    aim = pd.read_csv(aim_path)
    aim = aim.rename(columns={"symbol": "symbol"})
    aim["source_list"] = "AIM"
    aim = aim[["symbol", "company_name", "source_list"]]

    ftse = pd.read_csv(ftse_path)
    ftse = ftse.rename(columns={"ticker": "symbol"})
    ftse["source_list"] = "FTSE_SMALLCAP"
    ftse = ftse[["symbol", "company_name", "source_list"]]

    combined = pd.concat([aim, ftse], ignore_index=True)
    combined["symbol"] = combined["symbol"].astype(str).str.strip().str.upper()
    combined["company_name"] = combined["company_name"].astype(str).str.strip()

    dup_mask = combined.duplicated(subset="symbol", keep=False)
    if dup_mask.any():
        dupes = sorted(combined.loc[dup_mask, "symbol"].unique())
        print(f"[WARN] {len(dupes)} symbol(s) appear in both lists: {', '.join(dupes)}")

    combined = combined.drop_duplicates(subset="symbol", keep="first").reset_index(drop=True)
    return combined


def connect_ib(host, port, client_id, timeout=10):
    ib = IB()
    print(f"Connecting to TWS/IB Gateway at {host}:{port} (clientId={client_id})...")
    try:
        # readonly=True: this script only reads contract/market data, never trades.
        ib.connect(host, port, clientId=client_id, timeout=timeout, readonly=True)
    except Exception as exc:
        print("\n" + "=" * 70)
        print("Could not connect to TWS / IB Gateway.")
        print("Before running this script:")
        print("  1. Start TWS or IB Gateway and log in (paper or live).")
        print("  2. Configuration -> API -> Settings -> Enable ActiveX and Socket Clients.")
        print(f"  3. Confirm it is listening on port {port}.")
        print(f"     {DEFAULT_PORT_HINTS}")
        print("  4. Re-run this script.")
        print("=" * 70)
        raise SystemExit(1) from exc
    print("Connected.")
    return ib


def resolve_contract(ib, symbol, pacing_guard, throttle_sec):
    for attempt in EXCHANGE_ATTEMPTS:
        contract = Stock(
            symbol,
            attempt["exchange"],
            "GBP",
            primaryExchange=attempt["primaryExchange"],
        )
        try:
            details = ib.reqContractDetails(contract)
        except Exception:
            details = []
        ib.sleep(throttle_sec)
        pacing_guard.check_and_cooldown(ib, cooldown_sec=90)
        if details:
            chosen = next(
                (d for d in details if d.contract.primaryExchange == "LSE"),
                details[0],
            )
            return chosen
    return None


def run_contract_pass(ib, candidates_df, throttle_sec, checkpoint_path):
    pacing_guard = PacingGuard(ib)
    results = []
    total = len(candidates_df)
    try:
        for i, (_, row) in enumerate(candidates_df.iterrows()):
            symbol = row["symbol"]
            detail = resolve_contract(ib, symbol, pacing_guard, throttle_sec)
            if detail is not None:
                results.append(
                    {
                        "symbol": symbol,
                        "company_name": row["company_name"],
                        "source_list": row["source_list"],
                        "ibkr_tradable": True,
                        "ibkr_conid": detail.contract.conId,
                        "exchange": detail.contract.primaryExchange or detail.contract.exchange,
                        "avg_daily_volume": None,
                        "reason_dropped": "",
                    }
                )
            else:
                results.append(
                    {
                        "symbol": symbol,
                        "company_name": row["company_name"],
                        "source_list": row["source_list"],
                        "ibkr_tradable": False,
                        "ibkr_conid": None,
                        "exchange": None,
                        "avg_daily_volume": None,
                        "reason_dropped": "no_contract_found",
                    }
                )
            if (i + 1) % 25 == 0 or (i + 1) == total:
                print(f"  [{i + 1}/{total}] contract pass progress...")
                pd.DataFrame(results).to_csv(checkpoint_path, index=False)
    except KeyboardInterrupt:
        pd.DataFrame(results).to_csv(checkpoint_path, index=False)
        print(f"\nInterrupted. Checkpoint saved to {checkpoint_path} ({len(results)} rows).")
        raise
    return pd.DataFrame(results)


def fetch_adv(ib, symbol, conid, limiter, pacing_guard, duration="3 M", bar_size="1 day"):
    limiter.wait_if_needed(ib)
    contract = Contract(conId=int(conid), exchange="SMART", currency="GBP", symbol=symbol, secType="STK")
    try:
        bars = ib.reqHistoricalData(
            contract,
            endDateTime="",
            durationStr=duration,
            barSizeSetting=bar_size,
            whatToShow="TRADES",
            useRTH=True,
            formatDate=1,
        )
    except Exception as exc:
        return None, f"historical_data_error: {exc}"
    pacing_guard.check_and_cooldown(ib, cooldown_sec=120)
    if not bars:
        return None, "no_historical_bars"
    # barCount == 0 on every bar means IBKR returned synthetic/indicative closes with
    # no real trade prints behind them (seen for some SETSqx-quoted AIM names) -- the
    # contract is real but this ADV figure would be fabricated, so flag it rather than
    # silently reporting a misleadingly "verified" zero.
    if not any((b.barCount or 0) > 0 for b in bars):
        return None, "zero_trade_count_unverified"
    volumes = [b.volume for b in bars if b.volume is not None and b.volume >= 0]
    if not volumes:
        return None, "no_volume_data"
    return sum(volumes) / len(volumes), None


def run_liquidity_pass(ib, pass1_df, throttle_sec, checkpoint_path, max_calls, window_sec):
    limiter = SlidingWindowLimiter(max_calls=max_calls, period_sec=window_sec)
    pacing_guard = PacingGuard(ib)

    survivors = pass1_df[pass1_df["ibkr_tradable"] == True].copy()  # noqa: E712
    dropped = pass1_df[pass1_df["ibkr_tradable"] != True].copy()  # noqa: E712

    results = []
    total = len(survivors)
    try:
        for i, (_, row) in enumerate(survivors.iterrows()):
            adv, err = fetch_adv(ib, row["symbol"], row["ibkr_conid"], limiter, pacing_guard)
            rec = row.to_dict()
            rec["avg_daily_volume"] = adv
            rec["reason_dropped"] = err or ""
            results.append(rec)
            ib.sleep(throttle_sec)
            if (i + 1) % 20 == 0 or (i + 1) == total:
                print(f"  [{i + 1}/{total}] liquidity pass progress...")
                pd.concat([pd.DataFrame(results), dropped], ignore_index=True).to_csv(
                    checkpoint_path, index=False
                )
    except KeyboardInterrupt:
        pd.concat([pd.DataFrame(results), dropped], ignore_index=True).to_csv(
            checkpoint_path, index=False
        )
        print(f"\nInterrupted. Checkpoint saved to {checkpoint_path}.")
        raise

    survivors_df = pd.DataFrame(results) if results else survivors
    return pd.concat([survivors_df, dropped], ignore_index=True)


def parse_args():
    p = argparse.ArgumentParser(description="IB Ireland tradability + liquidity gate")
    p.add_argument("--aim-csv", default="aim-full-list-2026-08-13.csv")
    p.add_argument("--ftse-csv", default="ftse-smallcap-constituents-2026-08-13.csv")
    p.add_argument("--output", default="ibkr-tradability-gate.csv")
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument(
        "--port",
        type=int,
        default=7497,
        help=f"TWS/Gateway API port. {DEFAULT_PORT_HINTS}",
    )
    p.add_argument("--client-id", type=int, default=17)
    p.add_argument(
        "--contract-throttle",
        type=float,
        default=0.75,
        help="seconds to sleep between reqContractDetails calls (Pass 1)",
    )
    p.add_argument(
        "--hist-throttle",
        type=float,
        default=1.0,
        help="extra seconds to sleep between reqHistoricalData calls, on top of the rate limiter",
    )
    p.add_argument(
        "--hist-max-calls",
        type=int,
        default=55,
        help="reqHistoricalData calls allowed per --hist-window-sec (kept under IB's 60/10min limit)",
    )
    p.add_argument("--hist-window-sec", type=int, default=600)
    p.add_argument("--limit", type=int, default=None, help="only process first N candidates (testing)")
    p.add_argument(
        "--resume",
        action="store_true",
        help="resume Pass 1 from the existing checkpoint file, skipping already-processed symbols",
    )
    p.add_argument(
        "--skip-liquidity",
        action="store_true",
        help="run the contract-existence pass only; skip Pass 2 (reqHistoricalData)",
    )
    return p.parse_args()


def main():
    args = parse_args()

    print("=" * 70)
    print("IB Ireland tradability + liquidity gate")
    print("=" * 70)
    print("Requires TWS or IB Gateway running locally with the API enabled")
    print("(Configuration -> API -> Settings -> Enable ActiveX and Socket Clients).")
    print(f"  Target: {args.host}:{args.port}")
    print(f"  {DEFAULT_PORT_HINTS}")
    print("=" * 70)

    candidates = load_candidates(args.aim_csv, args.ftse_csv)
    n_aim = (candidates["source_list"] == "AIM").sum()
    n_ftse = (candidates["source_list"] == "FTSE_SMALLCAP").sum()
    print(f"Loaded {len(candidates)} unique candidate symbols ({n_aim} AIM, {n_ftse} FTSE SmallCap).")

    if args.limit:
        candidates = candidates.head(args.limit)
        print(f"--limit set: processing only first {len(candidates)} rows.")

    checkpoint1_path = Path(args.output).with_suffix(".pass1_checkpoint.csv")
    checkpoint2_path = Path(args.output).with_suffix(".pass2_checkpoint.csv")

    if args.resume and checkpoint1_path.exists():
        done = pd.read_csv(checkpoint1_path)
        done_symbols = set(done["symbol"])
        remaining = candidates[~candidates["symbol"].isin(done_symbols)]
        print(f"--resume set: {len(done_symbols)} symbols already in checkpoint, {len(remaining)} remaining.")
    else:
        done = pd.DataFrame()
        remaining = candidates

    ib = connect_ib(args.host, args.port, args.client_id)

    try:
        if len(remaining) > 0:
            new_results = run_contract_pass(ib, remaining, args.contract_throttle, checkpoint1_path)
            pass1_df = pd.concat([done, new_results], ignore_index=True) if len(done) else new_results
        else:
            pass1_df = done
        pass1_df.to_csv(checkpoint1_path, index=False)

        n_tradable = int(pass1_df["ibkr_tradable"].sum())
        print(f"Pass 1 complete: {n_tradable}/{len(pass1_df)} symbols have a tradable IB contract.")

        if args.skip_liquidity:
            final_df = pass1_df
        else:
            final_df = run_liquidity_pass(
                ib,
                pass1_df,
                args.hist_throttle,
                checkpoint2_path,
                args.hist_max_calls,
                args.hist_window_sec,
            )
    finally:
        ib.disconnect()

    final_df["ibkr_conid"] = final_df["ibkr_conid"].astype("Int64")
    final_df = final_df[
        [
            "symbol",
            "company_name",
            "source_list",
            "ibkr_tradable",
            "ibkr_conid",
            "exchange",
            "avg_daily_volume",
            "reason_dropped",
        ]
    ]
    final_df.to_csv(args.output, index=False)
    print(f"\nWrote {len(final_df)} rows to {args.output}")

    n_with_adv = final_df["avg_daily_volume"].notna().sum()
    print(f"  {n_with_adv} symbols have an average daily volume figure.")
    if (final_df["reason_dropped"] != "").any():
        print("  Drop reasons:")
        for reason, count in final_df.loc[final_df["reason_dropped"] != "", "reason_dropped"].value_counts().items():
            print(f"    {reason}: {count}")


if __name__ == "__main__":
    try:
        main()
    except SystemExit:
        raise
    except KeyboardInterrupt:
        print("\nAborted by user.")
        sys.exit(130)

#!/usr/bin/env python3
"""
Companies House ownership/control lookup for IBKR-tradable names.

Reads the IBKR tradability gate output, keeps only ibkr_tradable == True rows,
dedupes share classes of the same company (e.g. BMT/BMTO -> Braime Group PLC),
then resolves each company against the Companies House Public Data API and
pulls status, jurisdiction, officers, persons with significant control (PSC),
and registered charges.

API key
-------
Register for a free key at https://developer.company-information.service.gov.uk/
then either set it as an environment variable before running:

    $env:COMPANIES_HOUSE_API_KEY = "your-key-here"

or put it in a .env file (same directory as this script, not committed/shared
anywhere) as a single line:

    COMPANIES_HOUSE_API_KEY=your-key-here

The .env file is only read as a fallback when the environment variable isn't
already set. The key is HTTP Basic Auth username with a blank password --
never put it in a URL or log line.

Non-UK incorporation is expected, not an error
------------------------------------------------
Many AIM companies are incorporated in Jersey, Guernsey, Isle of Man, Bermuda,
Ireland, or the BVI. Those return no Companies House match correctly, because
they're genuinely outside its jurisdiction. This script distinguishes:
  - no_ch_match_likely_non_uk: zero search results, and nothing in the company
    name suggests we should have expected a UK entity -- expected outcome.
  - no_ch_match_unresolved: zero search results but the name carries a signal
    that made a UK match plausible (e.g. ends in "plc") -- worth a manual glance.
Neither is guessed at. Ambiguous searches with multiple similarly-plausible
candidates are logged as multiple_candidates and never auto-resolved to the
top hit -- a wrong company-number match silently poisons every downstream
field, which is worse than no data.
"""

import argparse
import difflib
import os
import re
import sys
import time
from collections import deque
from pathlib import Path

import pandas as pd
import requests

BASE_URL = "https://api.company-information.service.gov.uk"

# Suffixes stripped for dedup keys and for the secondary ("suffix-free") search
# attempt. Order doesn't matter -- stripped iteratively from the end of the
# tokenized name until none remain.
LEGAL_SUFFIXES = {
    "plc", "public", "limited", "ltd", "llp", "llc", "group", "holdings",
    "holding", "incorporated", "inc", "corporation", "corp", "company", "co",
    "sa", "nv", "ag", "se", "spa", "asa", "ab", "gmbh",
}

# Signals in the raw company name that suggest incorporation outside the UK,
# used only to classify an already-failed search, never to skip a search.
FOREIGN_NAME_MARKERS = re.compile(
    r"\((china|india|hong kong|bermuda|jersey|guernsey|isle of man|cayman|"
    r"bvi|british virgin islands|ireland|usa|canada|australia|south africa|"
    r"singapore|luxembourg|netherlands|switzerland)\)"
    r"|\b(inc|incorporated|corp|corporation|s\.a\.|n\.v\.|gmbh|s\.p\.a\.|pte)\.?\b",
    re.IGNORECASE,
)

# Match-confidence thresholds for scoring search results against the input name.
MATCH_THRESHOLD = 0.85
MATCH_GAP = 0.10
AMBIGUOUS_THRESHOLD = 0.55


class SlidingWindowLimiter:
    """Caps calls to max_calls within a trailing period_sec window."""

    def __init__(self, max_calls, period_sec):
        self.max_calls = max_calls
        self.period_sec = period_sec
        self.calls = deque()

    def wait_if_needed(self):
        now = time.monotonic()
        while self.calls and now - self.calls[0] > self.period_sec:
            self.calls.popleft()
        if len(self.calls) >= self.max_calls:
            sleep_for = self.period_sec - (now - self.calls[0]) + 1
            print(f"  [RATE LIMIT] at {self.max_calls}/{self.period_sec}s cap, sleeping {sleep_for:.0f}s")
            time.sleep(sleep_for)
        self.calls.append(time.monotonic())


class CompaniesHouseClient:
    def __init__(self, api_key, throttle_sec=0.3, max_calls=550, window_sec=300):
        self.session = requests.Session()
        self.session.auth = (api_key, "")
        self.throttle_sec = throttle_sec
        self.limiter = SlidingWindowLimiter(max_calls, window_sec)

    def get(self, path, params=None):
        self.limiter.wait_if_needed()
        url = f"{BASE_URL}{path}"
        for attempt in range(3):
            resp = self.session.get(url, params=params, timeout=15)
            time.sleep(self.throttle_sec)
            if resp.status_code == 200:
                return resp.json()
            if resp.status_code == 404:
                return None
            if resp.status_code == 401:
                raise SystemExit(
                    "\nCompanies House API returned 401 Unauthorized -- check that "
                    "COMPANIES_HOUSE_API_KEY is set to a valid key from "
                    "https://developer.company-information.service.gov.uk/\n"
                )
            if resp.status_code == 429:
                retry_after = int(resp.headers.get("Retry-After", 30))
                print(f"  [429] rate limited by CH, sleeping {retry_after}s")
                time.sleep(retry_after)
                continue
            if 500 <= resp.status_code < 600:
                backoff = 5 * (attempt + 1)
                print(f"  [{resp.status_code}] server error on {path}, retrying in {backoff}s")
                time.sleep(backoff)
                continue
            print(f"  [WARN] unexpected status {resp.status_code} on {path}: {resp.text[:200]}")
            return None
        print(f"  [WARN] giving up on {path} after retries")
        return None


def load_dotenv_key(key_name, path=".env"):
    """Minimal .env reader: returns key_name's value if the file has it, else None."""
    env_path = Path(path)
    if not env_path.exists():
        return None
    for line in env_path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        if k.strip() == key_name:
            return v.strip().strip('"').strip("'")
    return None


def strip_legal_suffixes(name):
    tokens = re.sub(r"[^\w\s&]", " ", name.lower()).split()
    while tokens and tokens[-1] in LEGAL_SUFFIXES:
        tokens.pop()
    return " ".join(tokens)


def normalize_for_dedup(name):
    return strip_legal_suffixes(name)


def name_similarity(a, b):
    return difflib.SequenceMatcher(None, strip_legal_suffixes(a), strip_legal_suffixes(b)).ratio()


def load_and_dedupe(input_path):
    df = pd.read_csv(input_path)
    tradable = df[df["ibkr_tradable"] == True].copy()  # noqa: E712
    tradable["dedup_key"] = tradable["company_name"].apply(normalize_for_dedup)

    kept_rows = []
    for key, group in tradable.groupby("dedup_key", sort=False):
        primary = group.iloc[0].copy()
        other_symbols = list(group.iloc[1:]["symbol"])
        primary["duplicate_tickers"] = ";".join(other_symbols) if other_symbols else ""
        kept_rows.append(primary)

    deduped = pd.DataFrame(kept_rows).drop(columns=["dedup_key"]).reset_index(drop=True)
    return deduped


def classify_no_match(company_name):
    if FOREIGN_NAME_MARKERS.search(company_name):
        return "no_ch_match_likely_non_uk"
    if strip_legal_suffixes(company_name).split() and company_name.strip().lower().endswith("plc"):
        return "no_ch_match_unresolved"
    return "no_ch_match_likely_non_uk"


def search_company(client, company_name):
    data = client.get("/search/companies", params={"q": company_name, "items_per_page": 5})
    items = (data or {}).get("items", [])
    if not items:
        stripped = strip_legal_suffixes(company_name)
        if stripped and stripped != company_name.lower():
            data = client.get("/search/companies", params={"q": stripped, "items_per_page": 5})
            items = (data or {}).get("items", [])
    return items


def light_normalize(name):
    """Case/punctuation-only normalization -- no suffix stripping, so distinct
    legal entities (e.g. a Holdings Plc vs its operating LLP) don't collapse
    into the same string the way the aggressive dedup normalization would."""
    return re.sub(r"\s+", " ", re.sub(r"[^\w\s]", "", name.lower())).strip()


def resolve_company(client, company_name):
    items = search_company(client, company_name)
    if not items:
        return classify_no_match(company_name), None, None, []

    # Exact match modulo case/punctuation against an active company is about as
    # confident as this gets -- short-circuit before the fuzzy suffix-stripped
    # scoring below, which can score two genuinely different but suffix-similar
    # entities (e.g. "...Holdings Plc" vs "...LLP") as identical and falsely
    # flag an unambiguous case as multiple_candidates.
    input_light = light_normalize(company_name)
    for it in items:
        if light_normalize(it.get("title", "")) == input_light and it.get("company_status") == "active":
            return "matched", it.get("company_number"), it.get("title"), []

    scored = sorted(
        ((name_similarity(company_name, it.get("title", "")), it) for it in items),
        key=lambda t: t[0],
        reverse=True,
    )
    top_score, top_item = scored[0]
    second_score = scored[1][0] if len(scored) > 1 else 0.0
    top_is_active = top_item.get("company_status") == "active"

    # A dissolved/closed top hit is never auto-matched, no matter how high the
    # string score -- generic words (e.g. "yellowcake" as a uranium industry
    # term) can make an unrelated defunct shell score deceptively high against
    # a currently-active, currently-tradable company. Fall through to manual
    # review instead of silently attaching the wrong company_number.
    if top_score >= MATCH_THRESHOLD and (top_score - second_score) >= MATCH_GAP and top_is_active:
        return "matched", top_item.get("company_number"), top_item.get("title"), []

    plausible = [it for score, it in scored if score >= AMBIGUOUS_THRESHOLD]
    if len(plausible) >= 1:
        candidates = [
            f"{it.get('company_number')}:{it.get('title')} [{it.get('company_status')}]" for _, it in scored[:3]
        ]
        return "multiple_candidates", None, None, candidates

    return classify_no_match(company_name), None, None, []


def fetch_profile(client, company_number):
    return client.get(f"/company/{company_number}") or {}


def fetch_officers_count(client, company_number):
    data = client.get(f"/company/{company_number}/officers", params={"items_per_page": 100})
    if not data:
        return 0
    items = data.get("items", [])
    current_directors = [
        it for it in items
        if it.get("officer_role") == "director" and not it.get("resigned_on")
    ]
    return len(current_directors)


PSC_BAND_RE = re.compile(r"(\d+)-to-(\d+)-percent")
PSC_STATEMENT_KIND = "persons-with-significant-control-statement"


def fetch_psc(client, company_number):
    data = client.get(f"/company/{company_number}/persons-with-significant-control", params={"items_per_page": 100})
    if not data:
        return 0, ""
    items = data.get("items", [])
    real_psc = [it for it in items if it.get("kind") != PSC_STATEMENT_KIND]

    best_band = None
    best_upper = -1
    for it in real_psc:
        for noc in it.get("natures_of_control", []):
            m = PSC_BAND_RE.search(noc)
            if m:
                upper = int(m.group(2))
                if upper > best_upper:
                    best_upper = upper
                    best_band = f"{m.group(1)}-{m.group(2)}%"

    return len(real_psc), (best_band or "")


def fetch_charges(client, company_number):
    data = client.get(f"/company/{company_number}/charges", params={"items_per_page": 100})
    if not data:
        return 0, 0
    items = data.get("items", [])
    total = data.get("total_count", len(items))
    outstanding = sum(1 for it in items if it.get("status") == "outstanding")
    return total, outstanding


def run_lookup(client, companies_df, checkpoint_path):
    results = []
    total = len(companies_df)
    try:
        for i, (_, row) in enumerate(companies_df.iterrows()):
            symbol = row["symbol"]
            company_name = row["company_name"]
            status, company_number, registered_name, candidates = resolve_company(client, company_name)

            rec = {
                "symbol": symbol,
                "company_name_input": company_name,
                "duplicate_tickers": row.get("duplicate_tickers", ""),
                "ch_match_status": status,
                "ch_company_number": company_number,
                "ch_registered_name": registered_name,
                "ch_status": None,
                "jurisdiction": None,
                "incorporation_date": None,
                "sic_codes": None,
                "num_officers": None,
                "num_psc": None,
                "max_psc_ownership_band": None,
                "num_charges": None,
                "num_outstanding_charges": None,
            }

            if status == "multiple_candidates":
                rec["ch_registered_name"] = " | ".join(candidates)

            if status == "matched" and company_number:
                profile = fetch_profile(client, company_number)
                rec["ch_status"] = profile.get("company_status")
                rec["jurisdiction"] = profile.get("jurisdiction")
                rec["incorporation_date"] = profile.get("date_of_creation")
                rec["sic_codes"] = ";".join(profile.get("sic_codes", []) or [])
                rec["num_officers"] = fetch_officers_count(client, company_number)
                num_psc, band = fetch_psc(client, company_number)
                rec["num_psc"] = num_psc
                rec["max_psc_ownership_band"] = band
                total_charges, outstanding_charges = fetch_charges(client, company_number)
                rec["num_charges"] = total_charges
                rec["num_outstanding_charges"] = outstanding_charges

            results.append(rec)

            if (i + 1) % 25 == 0 or (i + 1) == total:
                print(f"  [{i + 1}/{total}] lookup progress...")
                pd.DataFrame(results).to_csv(checkpoint_path, index=False)
    except KeyboardInterrupt:
        pd.DataFrame(results).to_csv(checkpoint_path, index=False)
        print(f"\nInterrupted. Checkpoint saved to {checkpoint_path} ({len(results)} rows).")
        raise
    return pd.DataFrame(results)


def parse_args():
    p = argparse.ArgumentParser(description="Companies House lookup for IBKR-tradable names")
    p.add_argument("--input", default="ibkr-tradability-gate-v2.csv")
    p.add_argument("--output", default="companies-house-lookup.csv")
    p.add_argument("--api-key", default=None, help="overrides COMPANIES_HOUSE_API_KEY env var")
    p.add_argument("--throttle", type=float, default=0.3, help="seconds to sleep between CH API calls")
    p.add_argument("--max-calls", type=int, default=550, help="calls allowed per --window-sec (kept under CH's 600/5min)")
    p.add_argument("--window-sec", type=int, default=300)
    p.add_argument("--limit", type=int, default=None, help="only process first N companies (testing)")
    p.add_argument("--resume", action="store_true")
    return p.parse_args()


def main():
    args = parse_args()
    api_key = args.api_key or os.environ.get("COMPANIES_HOUSE_API_KEY") or load_dotenv_key("COMPANIES_HOUSE_API_KEY")
    if not api_key:
        print("=" * 70)
        print("No Companies House API key found.")
        print("Register at https://developer.company-information.service.gov.uk/")
        print("then either set it before running:")
        print('  $env:COMPANIES_HOUSE_API_KEY = "your-key-here"')
        print("or create a .env file in this directory containing:")
        print("  COMPANIES_HOUSE_API_KEY=your-key-here")
        print("or pass --api-key directly.")
        print("=" * 70)
        raise SystemExit(1)

    companies = load_and_dedupe(args.input)
    print(f"Loaded {len(companies)} unique companies after dedup (from ibkr_tradable rows in {args.input}).")
    dupes = companies[companies["duplicate_tickers"] != ""]
    if len(dupes):
        print(f"  {len(dupes)} companies had duplicate share-class tickers folded in:")
        for _, r in dupes.iterrows():
            print(f"    {r['symbol']} (kept) + {r['duplicate_tickers']} -> {r['company_name']}")

    if args.limit:
        companies = companies.head(args.limit)
        print(f"--limit set: processing only first {len(companies)} rows.")

    checkpoint_path = Path(args.output).with_suffix(".checkpoint.csv")

    if args.resume and checkpoint_path.exists():
        done = pd.read_csv(checkpoint_path)
        done_symbols = set(done["symbol"])
        remaining = companies[~companies["symbol"].isin(done_symbols)]
        print(f"--resume set: {len(done_symbols)} already in checkpoint, {len(remaining)} remaining.")
    else:
        done = pd.DataFrame()
        remaining = companies

    client = CompaniesHouseClient(api_key, throttle_sec=args.throttle, max_calls=args.max_calls, window_sec=args.window_sec)

    if len(remaining) > 0:
        new_results = run_lookup(client, remaining, checkpoint_path)
        final_df = pd.concat([done, new_results], ignore_index=True) if len(done) else new_results
    else:
        final_df = done

    final_df = final_df[
        [
            "symbol",
            "company_name_input",
            "duplicate_tickers",
            "ch_match_status",
            "ch_company_number",
            "ch_registered_name",
            "ch_status",
            "jurisdiction",
            "incorporation_date",
            "sic_codes",
            "num_officers",
            "num_psc",
            "max_psc_ownership_band",
            "num_charges",
            "num_outstanding_charges",
        ]
    ]
    final_df.to_csv(args.output, index=False)
    print(f"\nWrote {len(final_df)} rows to {args.output}")
    print(final_df["ch_match_status"].value_counts())


if __name__ == "__main__":
    try:
        main()
    except SystemExit:
        raise
    except KeyboardInterrupt:
        print("\nAborted by user.")
        sys.exit(130)

#!/usr/bin/env python3
"""
Resolve the multiple_candidates rows from companies_house_lookup.py that
actually have a single, name-confident answer among their candidate list.

A row qualifies if, after normalizing candidate titles and the input company
name (uppercase, strip punctuation, strip trailing PLC/LIMITED/LTD), exactly
one candidate's name matches the input -- or, if two candidates match (e.g. a
UK-registered entity and its overseas-company branch/establishment register
entry, or an active/dissolved pair), exactly one of them has company_status
== "active". Either way there is exactly one sane choice, so this is a
name-confidence bar, not a guess.

Everything else -- genuinely ambiguous generic names, and rows that never had
a real candidate to begin with (the HUTCHMED/Yellow Cake pattern) -- is left
untouched. Per the same "don't spend effort until it survives the next
filter" discipline used elsewhere in this pipeline, those are deferred to
manual lookup only if they survive later filtering.

Qualifying rows get promoted to ch_match_status = "matched_high_confidence"
and the same four profile calls already made for the clean "matched" rows
(company profile, officers, PSC, charges) are run against the resolved
company number. The main output CSV is updated in place; row order and every
other row are left untouched.
"""

import argparse
import os
import re
import sys
from pathlib import Path

import pandas as pd

from companies_house_lookup import (
    CompaniesHouseClient,
    fetch_charges,
    fetch_officers_count,
    fetch_profile,
    fetch_psc,
    load_dotenv_key,
)

CANDIDATE_RE = re.compile(r"^(?P<number>\S+):(?P<title>.*?)\s\[(?P<status>[^\]]*)\]$")
TRAILING_SUFFIXES = {"PLC", "LIMITED", "LTD"}


def normalize_exact(name):
    tokens = re.sub(r"[^\w\s]", " ", str(name).upper()).split()
    while tokens and tokens[-1] in TRAILING_SUFFIXES:
        tokens.pop()
    return " ".join(tokens)


def parse_candidates(field):
    out = []
    for part in str(field).split("|"):
        part = part.strip()
        m = CANDIDATE_RE.match(part)
        if m:
            out.append((m.group("number"), m.group("title"), m.group("status")))
    return out


def resolve_exact_candidate(company_name_input, candidate_field):
    candidates = parse_candidates(candidate_field)
    target = normalize_exact(company_name_input)
    matches = [c for c in candidates if normalize_exact(c[1]) == target]
    if len(matches) == 1:
        return matches[0]
    if len(matches) >= 2:
        actives = [c for c in matches if c[2] == "active"]
        if len(actives) == 1:
            return actives[0]
    return None


def main():
    p = argparse.ArgumentParser(description="Resolve name-confident multiple_candidates rows")
    p.add_argument("--input", default="companies-house-lookup.csv")
    p.add_argument("--api-key", default=None)
    p.add_argument("--throttle", type=float, default=0.3)
    args = p.parse_args()

    api_key = args.api_key or os.environ.get("COMPANIES_HOUSE_API_KEY") or load_dotenv_key("COMPANIES_HOUSE_API_KEY")
    if not api_key:
        print("No Companies House API key found (env var or .env). Aborting.")
        raise SystemExit(1)

    df = pd.read_csv(args.input)
    mc_mask = df["ch_match_status"] == "multiple_candidates"
    print(f"multiple_candidates rows: {mc_mask.sum()}")

    to_resolve = []
    for idx in df[mc_mask].index:
        row = df.loc[idx]
        resolved = resolve_exact_candidate(row["company_name_input"], row["ch_registered_name"])
        if resolved:
            to_resolve.append((idx, resolved))

    print(f"Name-confident resolutions found: {len(to_resolve)}")
    if not to_resolve:
        print("Nothing to do.")
        return

    client = CompaniesHouseClient(api_key, throttle_sec=args.throttle)

    for i, (idx, (company_number, title, status)) in enumerate(to_resolve):
        symbol = df.loc[idx, "symbol"]
        print(f"  [{i + 1}/{len(to_resolve)}] {symbol} -> {company_number} ({title})")

        profile = fetch_profile(client, company_number)
        num_officers = fetch_officers_count(client, company_number)
        num_psc, band = fetch_psc(client, company_number)
        total_charges, outstanding_charges = fetch_charges(client, company_number)

        df.loc[idx, "ch_match_status"] = "matched_high_confidence"
        df.loc[idx, "ch_company_number"] = company_number
        df.loc[idx, "ch_registered_name"] = title
        df.loc[idx, "ch_status"] = profile.get("company_status")
        df.loc[idx, "jurisdiction"] = profile.get("jurisdiction")
        df.loc[idx, "incorporation_date"] = profile.get("date_of_creation")
        df.loc[idx, "sic_codes"] = ";".join(profile.get("sic_codes", []) or [])
        df.loc[idx, "num_officers"] = num_officers
        df.loc[idx, "num_psc"] = num_psc
        df.loc[idx, "max_psc_ownership_band"] = band
        df.loc[idx, "num_charges"] = total_charges
        df.loc[idx, "num_outstanding_charges"] = outstanding_charges

    df.to_csv(args.input, index=False)
    print(f"\nUpdated {len(to_resolve)} rows in {args.input}")
    print(df["ch_match_status"].value_counts())


if __name__ == "__main__":
    try:
        main()
    except SystemExit:
        raise
    except KeyboardInterrupt:
        print("\nAborted by user.")
        sys.exit(130)

#!/usr/bin/env python3
"""
FTSE SmallCap market cap pull from stockanalysis.com.

Primary path: scrape stockanalysis.com/list/london-stock-exchange/ pages 1-7
and exact-match tickers against the known FTSE SmallCap constituent list.
That list page turned out to only cover ~half of our 189 tickers in testing
-- it's a market-cap-ranked, truncated view dominated by thousands of
foreign cross-listings (NVIDIA, Apple, etc.), not a complete Main Market
roster, and it appears to exclude AIM-classified tickers entirely.

Fallback path: for anything not found on the list pages, fetch the
individual stockanalysis.com/quote/lon/{TICKER}/ page directly, which
reliably shows "Market Cap" for ordinary operating companies and REITs.

Hard limit, not a bug: genuine closed-end investment trusts/funds (heavily
represented in the FTSE SmallCap index) have no "Market Cap" stat anywhere
on this vendor -- their quote pages show "Fund Assets" instead, usually
"n/a". This script detects that pattern and logs it as
no_market_cap_fund_type, distinct from a genuine no_market_cap_not_found
miss, rather than silently leaving both blank with no explanation.

robots.txt for stockanalysis.com permits /list/ and /quote/ for a generic
user agent (checked before writing this script).
"""

import argparse
import re
import sys
import time
from pathlib import Path

import pandas as pd
import requests
from bs4 import BeautifulSoup

BASE_LIST_URL = "https://stockanalysis.com/list/london-stock-exchange/"
QUOTE_URL_TMPL = "https://stockanalysis.com/quote/lon/{ticker}/"
HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
NUM_LIST_PAGES = 7

CAP_MULTIPLIERS = {"T": 1e12, "B": 1e9, "M": 1e6, "K": 1e3}


def parse_market_cap(text):
    text = (text or "").strip()
    if not text or text in {"-", "n/a", "N/A"}:
        return None
    m = re.match(r"^([\d.]+)([TBMK])?$", text)
    if not m:
        return None
    value = float(m.group(1))
    return value * CAP_MULTIPLIERS.get(m.group(2), 1)


def http_get(session, url, throttle_sec, max_retries=3):
    for attempt in range(max_retries):
        try:
            resp = session.get(url, timeout=20)
        except requests.RequestException as exc:
            print(f"  [WARN] request error on {url}: {exc}")
            time.sleep(3 * (attempt + 1))
            continue
        time.sleep(throttle_sec)
        if resp.status_code == 200:
            return resp
        if resp.status_code == 404:
            return resp
        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", 15))
            print(f"  [429] rate limited on {url}, sleeping {retry_after}s")
            time.sleep(retry_after)
            continue
        if 500 <= resp.status_code < 600:
            backoff = 5 * (attempt + 1)
            print(f"  [{resp.status_code}] server error on {url}, retrying in {backoff}s")
            time.sleep(backoff)
            continue
        print(f"  [WARN] unexpected status {resp.status_code} on {url}")
        return resp
    print(f"  [WARN] giving up on {url} after retries")
    return None


def scrape_list_pages(session, throttle_sec, num_pages=NUM_LIST_PAGES):
    rows_by_symbol = {}
    for page in range(1, num_pages + 1):
        url = BASE_LIST_URL if page == 1 else f"{BASE_LIST_URL}?page={page}"
        resp = http_get(session, url, throttle_sec)
        if resp is None or resp.status_code != 200:
            print(f"  [WARN] failed to fetch list page {page}, skipping")
            continue
        soup = BeautifulSoup(resp.text, "lxml")
        table = soup.find("table", id="main-table")
        if table is None:
            print(f"  [WARN] no table found on list page {page}")
            continue
        trs = table.find("tbody").find_all("tr")
        for tr in trs:
            cells = [td.get_text(strip=True) for td in tr.find_all("td")]
            if len(cells) >= 4:
                symbol = cells[1]
                rows_by_symbol[symbol] = {
                    "stockanalysis_name": cells[2],
                    "market_cap_raw": cells[3],
                }
        print(f"  list page {page}/{num_pages}: {len(trs)} rows")
    return rows_by_symbol


def fetch_quote_page_market_cap(session, ticker, throttle_sec):
    url = QUOTE_URL_TMPL.format(ticker=ticker)
    resp = http_get(session, url, throttle_sec)
    if resp is None:
        return None, "no_market_cap_not_found", None
    if resp.status_code == 404:
        return None, "no_market_cap_not_found", None

    soup = BeautifulSoup(resp.text, "lxml")
    title_tag = soup.find("title")
    stockanalysis_name = title_tag.get_text() if title_tag else None
    text = soup.get_text(" ", strip=True)

    m = re.search(r"Market Cap\s*([\d.]+[TBMK]?|n/a)", text)
    if m and m.group(1) != "n/a":
        return m.group(1), "matched", stockanalysis_name

    # Fund/trust template variants: some show "Fund Assets", others just
    # "Assets" -- "Expense Ratio" is the marker that's consistently present
    # on every fund-type page and absent from operating-company pages.
    if "Expense Ratio" in text:
        return None, "no_market_cap_fund_type", stockanalysis_name

    return None, "no_market_cap_not_found", stockanalysis_name


def run(known_df, throttle_sec, checkpoint_path, num_pages, limit=None):
    session = requests.Session()
    session.headers.update(HEADERS)

    if limit:
        known_df = known_df.head(limit)

    print(f"Scraping {num_pages} list page(s)...")
    list_rows = scrape_list_pages(session, throttle_sec, num_pages)
    print(f"List pages yielded {len(list_rows)} total symbol rows.")

    results = []
    total = len(known_df)
    try:
        for i, (_, row) in enumerate(known_df.iterrows()):
            ticker = row["ticker"]
            company_name = row["company_name"]

            if ticker in list_rows:
                raw = list_rows[ticker]
                results.append(
                    {
                        "ticker": ticker,
                        "company_name": company_name,
                        "stockanalysis_name": raw["stockanalysis_name"],
                        "market_cap_gbp": parse_market_cap(raw["market_cap_raw"]),
                        "match_status": "matched",
                        "source": "list_page",
                    }
                )
            else:
                raw_cap, status, stockanalysis_name = fetch_quote_page_market_cap(session, ticker, throttle_sec)
                results.append(
                    {
                        "ticker": ticker,
                        "company_name": company_name,
                        "stockanalysis_name": stockanalysis_name,
                        "market_cap_gbp": parse_market_cap(raw_cap) if raw_cap else None,
                        "match_status": status,
                        "source": "quote_page" if status == "matched" else "",
                    }
                )

            if (i + 1) % 25 == 0 or (i + 1) == total:
                print(f"  [{i + 1}/{total}] lookup progress...")
                pd.DataFrame(results).to_csv(checkpoint_path, index=False)
    except KeyboardInterrupt:
        pd.DataFrame(results).to_csv(checkpoint_path, index=False)
        print(f"\nInterrupted. Checkpoint saved to {checkpoint_path} ({len(results)} rows).")
        raise

    return pd.DataFrame(results)


def parse_args():
    p = argparse.ArgumentParser(description="FTSE SmallCap market cap pull from stockanalysis.com")
    p.add_argument("--input", default="ftse-smallcap-constituents-2026-08-13.csv")
    p.add_argument("--output", default="ftse-smallcap-marketcap.csv")
    p.add_argument("--throttle", type=float, default=0.5, help="seconds to sleep between HTTP requests")
    p.add_argument("--num-pages", type=int, default=NUM_LIST_PAGES)
    p.add_argument("--limit", type=int, default=None, help="only process first N tickers (testing)")
    return p.parse_args()


def main():
    args = parse_args()
    known_df = pd.read_csv(args.input)
    print(f"Loaded {len(known_df)} known FTSE SmallCap tickers from {args.input}.")

    checkpoint_path = Path(args.output).with_suffix(".checkpoint.csv")
    final_df = run(known_df, args.throttle, checkpoint_path, args.num_pages, args.limit)

    final_df = final_df[["ticker", "company_name", "stockanalysis_name", "market_cap_gbp", "match_status", "source"]]
    final_df.to_csv(args.output, index=False)
    print(f"\nWrote {len(final_df)} rows to {args.output}")
    print(final_df["match_status"].value_counts())


if __name__ == "__main__":
    try:
        main()
    except SystemExit:
        raise
    except KeyboardInterrupt:
        print("\nAborted by user.")
        sys.exit(130)