Build Your Own Rank Tracker With Python and Proxies

The most common bug in homemade trackers is a single nullable position column. NULL then means two opposite things: "we checked and you're not in the top 20" and "we never got a valid page". Charted, both look like a drop. Store the verdict next to the position and the ambiguity disappears.

CREATE TABLE keywords (
  id       bigserial PRIMARY KEY,
  keyword  text     NOT NULL,
  country  char(2)  NOT NULL,
  lang     text     NOT NULL,
  domain   text     NOT NULL,
  depth    smallint NOT NULL DEFAULT 20,
  active   boolean  NOT NULL DEFAULT true,
  UNIQUE (keyword, country, lang, domain)
);

CREATE TABLE rank_checks (
  keyword_id  bigint      NOT NULL REFERENCES keywords(id),
  checked_on  date        NOT NULL,
  verdict     text        NOT NULL,  -- ok, consent, captcha, js_wall, empty, http_error
  position    smallint,              -- NULL with verdict 'ok' = not within depth
  url         text,                  -- which of your pages ranked
  route       text        NOT NULL,  -- proxy, api, api_js, api_consent, api_premium
  attempts    smallint    NOT NULL,
  credits     integer     NOT NULL DEFAULT 0,
  fetched_at  timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (keyword_id, checked_on)
);

CREATE INDEX rank_checks_day ON rank_checks (checked_on);

The upsert below makes the daily job safe to run twice. Its WHERE clause means a second pass can overwrite a failed row but never a good one, so a 09:00 retry run only touches the keywords that broke at 06:00.

# db.py
import psycopg

DSN = "postgresql://tracker:change-me@localhost:5432/ranks"


def sync_keywords(conn, keywords):
    ids = {}
    with conn.cursor() as cur:
        for k in keywords:
            cur.execute(
                """INSERT INTO keywords (keyword, country, lang, domain, depth)
                   VALUES (%s, %s, %s, %s, %s)
                   ON CONFLICT (keyword, country, lang, domain)
                   DO UPDATE SET depth = EXCLUDED.depth, active = true
                   RETURNING id""",
                (k.keyword, k.country, k.lang, k.domain, k.depth),
            )
            ids[k] = cur.fetchone()[0]
    conn.commit()
    return ids


def save_check(conn, keyword_id, day, verdict, position, url, route, attempts, credits):
    conn.execute(
        """INSERT INTO rank_checks
             (keyword_id, checked_on, verdict, position, url, route, attempts, credits)
           VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
           ON CONFLICT (keyword_id, checked_on) DO UPDATE SET
             verdict = EXCLUDED.verdict, position = EXCLUDED.position,
             url = EXCLUDED.url, route = EXCLUDED.route,
             attempts = rank_checks.attempts + EXCLUDED.attempts,
             credits = rank_checks.credits + EXCLUDED.credits,
             fetched_at = now()
           WHERE rank_checks.verdict <> 'ok'""",
        (keyword_id, day, verdict, position, url, route, attempts, credits),
    )
    conn.commit()

The attempts and credits columns accumulate across passes, so a monthly SUM(credits) grouped by keyword shows you which terms are expensive to measure. Those are the ones to move to weekly checks. For wider storage trade-offs (files versus SQLite versus a server database), see how to store scraped data.

Each keyword climbs a ladder from cheapest to most expensive and stops at the first rung that returns a real results page. The verdict decides the next rung: a captcha earns another proxy attempt on a new IP, a consent page or JavaScript wall skips straight past the routes that can't fix it.

Credit figures include the +5 for country_code, which every rung sends.

# run_daily.py
import datetime as dt
import random
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import psycopg
import requests

from classify import CONSENT_SCENARIO, Verdict, classify
from db import DSN, save_check, sync_keywords
from fetch_api import ApiThrottled, fetch_via_api
from fetch_proxy import fetch_page_via_proxy, new_session, serp_url
from keywords import load_keywords
from parse import find_position, organic_results

ROUTES = {
    "api": {"render_js": False},
    "api_js": {"render_js": True},
    "api_consent": {"render_js": True, "scenario": CONSENT_SCENARIO},
    "api_premium": {"render_js": True, "premium": True},
}
ORDER = ["proxy", "api", "api_js", "api_premium"]
MAX_ATTEMPTS = 6


def attempt(route, kw):
    results, credits = [], 0
    session = new_session(kw.country, kw.lang) if route == "proxy" else None
    for start in range(0, kw.depth, 10):
        url = serp_url(kw.keyword, kw.country, kw.lang, start)
        if session is not None:
            status, final_url, html = fetch_page_via_proxy(session, url)
        else:
            status, html, used = fetch_via_api(url, kw.country, **ROUTES[route])
            final_url, credits = None, credits + used
        page = organic_results(html)
        verdict = classify(status, final_url, html, len(page))
        if verdict is Verdict.EMPTY and start > 0:
            break  # the SERP simply ran out of results
        if verdict is not Verdict.OK:
            return verdict, results, credits
        results.extend(u for u in page if u not in results)
        time.sleep(random.uniform(3, 8))
    return Verdict.OK, results, credits


def next_route(route, verdict, tries_here):
    if verdict is Verdict.CONSENT:
        return None if route == "api_consent" else "api_consent"
    if verdict is Verdict.JS_WALL and route in ("proxy", "api"):
        return "api_js"
    if route == "proxy" and tries_here < 3:
        return "proxy"
    if route == "api_consent":
        return None
    i = ORDER.index(route)
    return ORDER[i + 1] if i + 1 < len(ORDER) else None


def check_keyword(kw):
    route, last, tries_here, attempts, credits = "proxy", "proxy", 0, 0, 0
    verdict, results = Verdict.HTTP_ERROR, []
    while route is not None and attempts < MAX_ATTEMPTS:
        attempts, tries_here, last = attempts + 1, tries_here + 1, route
        try:
            verdict, results, used = attempt(route, kw)
        except ApiThrottled as t:
            time.sleep(t.wait)  # a throttle is not a failed attempt
            attempts, tries_here = attempts - 1, tries_here - 1
            continue
        except requests.RequestException:
            verdict, results, used = Verdict.HTTP_ERROR, [], 0
        credits += used
        if verdict is Verdict.OK:
            break
        nxt = next_route(route, verdict, tries_here)
        tries_here = tries_here if nxt == route else 0
        route = nxt
    if verdict is Verdict.OK:
        position, url = find_position(results, kw.domain)
    else:
        position, url = None, None
    return kw, verdict, position, url, last, attempts, credits


def main(csv_path="keywords.csv", workers=8):
    today = dt.date.today()
    keywords = load_keywords(csv_path)
    with psycopg.connect(DSN) as conn:
        ids = sync_keywords(conn, keywords)
        done = {row[0] for row in conn.execute(
            "SELECT keyword_id FROM rank_checks WHERE checked_on = %s AND verdict = 'ok'",
            (today,))}
        todo = [k for k in keywords if ids[k] not in done]
        with ThreadPoolExecutor(max_workers=workers) as pool:
            for fut in as_completed([pool.submit(check_keyword, k) for k in todo]):
                kw, verdict, pos, url, route, n, cr = fut.result()
                save_check(conn, ids[kw], today, verdict.value, pos, url, route, n, cr)
                print(f"{verdict.value:<10} {pos or '-':>4}  {route:<11} {kw.keyword}")


if __name__ == "__main__":
    main()

Workers only fetch; the main thread does every database write, so one connection is enough and you never share a psycopg connection across threads. The skip list at the top of main is what makes the second pass cheap: keywords that already have an ok row today aren't fetched again.

Schedule two passes. The first does the work; the second mops up whatever failed, after Google has had a few hours to forget about any IPs that got flagged.

# crontab -e   (server clock in UTC)
15 6 * * *  cd /opt/rank-tracker && .venv/bin/python run_daily.py >> logs/daily.log 2>&1
15 9 * * *  cd /opt/rank-tracker && .venv/bin/python run_daily.py >> logs/daily.log 2>&1
30 7 * * 1  cd /opt/rank-tracker && .venv/bin/python report.py > reports/$(date +\%F).md

Keep the run time fixed. Rankings shift through the day, and a check that drifts from 06:15 to 14:00 adds noise you'll later mistake for movement. Locking, alerting and systemd timers as an alternative to cron are covered in how to schedule and automate web scrapers.

Comparing today with exactly seven days ago is noisy. One odd morning on either side produces a big swing that isn't there the next day. Compare the median of the last three good checks with the median of three good checks a week earlier instead, and treat "not in top N" as depth plus one so entries and drops still produce a number.

-- weekly.sql
WITH windowed AS (
  SELECT c.keyword_id,
         CASE WHEN c.checked_on > %(today)s - 3 THEN 'now' ELSE 'then' END AS bucket,
         COALESCE(c.position, k.depth + 1) AS pos,
         c.url
  FROM rank_checks c
  JOIN keywords k ON k.id = c.keyword_id
  WHERE c.verdict = 'ok'
    AND (c.checked_on > %(today)s - 3
         OR c.checked_on BETWEEN %(today)s - 9 AND %(today)s - 7)
), summary AS (
  SELECT keyword_id, bucket,
         percentile_disc(0.5) WITHIN GROUP (ORDER BY pos) AS pos,
         mode() WITHIN GROUP (ORDER BY url) AS url
  FROM windowed
  GROUP BY keyword_id, bucket
)
SELECT k.keyword, k.country, t.pos AS was, n.pos AS now,
       t.pos - n.pos AS gain,
       n.url IS DISTINCT FROM t.url AS url_changed,
       k.depth
FROM summary n
JOIN summary t ON t.keyword_id = n.keyword_id AND t.bucket = 'then'
JOIN keywords k ON k.id = n.keyword_id
WHERE n.bucket = 'now'
ORDER BY gain DESC;

Because the query only reads verdict = 'ok' rows, a morning of captchas lowers the sample size instead of inventing a drop. A keyword with no good checks in either window just doesn't appear, which is the honest outcome.

# report.py
import datetime as dt
from pathlib import Path

import psycopg

from db import DSN

REPORT_SQL = Path("weekly.sql").read_text(encoding="utf-8")


def label(pos: int, depth: int) -> str:
    return "-" if pos > depth else str(pos)


def weekly_report(today=None, threshold=3):
    today = today or dt.date.today()
    with psycopg.connect(DSN) as conn:
        rows = conn.execute(REPORT_SQL, {"today": today}).fetchall()
    groups = {"Entered": [], "Dropped out": [], "Gained": [], "Lost": [],
              "Ranking URL changed": []}
    for kw, country, was, now, gain, url_changed, depth in rows:
        line = f"| {kw} | {country} | {label(was, depth)} | {label(now, depth)} | {gain:+d} |"
        if was > depth >= now:
            groups["Entered"].append(line)
        elif now > depth >= was:
            groups["Dropped out"].append(line)
        elif gain >= threshold:
            groups["Gained"].append(line)
        elif gain <= -threshold:
            groups["Lost"].append(line)
        if url_changed and now <= depth and was <= depth:
            groups["Ranking URL changed"].append(line)
    out = [f"Rank movement, week to {today:%Y-%m-%d}", ""]
    for title, lines in groups.items():
        if lines:
            out += [f"{title} ({len(lines)})", "",
                    "| Keyword | Country | Was | Now | Change |",
                    "|---|---|---|---|---|", *lines, ""]
    return "\n".join(out)


if __name__ == "__main__":
    print(weekly_report())

The "Ranking URL changed" group is the one clients don't get from most dashboards. When the ranking page flips from a guide to a category page, the position may barely move, but it often means two of their pages are competing for one query. It's a content conversation, and you bring the evidence.

Prices below are in US dollars, excluding VAT, read from sparkproxy.io on 2026-10-30. Proxy plans are packs billed by threads, not by traffic, and bandwidth is unlimited: Starter $75/month (100 threads), Core $140 (250), Boost $240 (500), Plus $440 (1,000). Longer terms take 5%, 10% or 15% off for 3, 6 or 12 months, and there's a 24-hour free trial with no credit card. Scraping API plans are credit packs: Starter $49/month for 250K credits, Growth $99 for 1M, Pro $249 for 3M, Scale $599 for 8M, and new accounts get 1,000 free credits.

At depth 20, 1,000 keywords is 2,000 page fetches a day, or about 60,000 a month.

Nobody should run everything on the bottom row. The ladder's real cost is the proxy plan plus credits for the keywords that escalate, and that's easy to estimate: every 1% of keywords that land on rung 3 adds 10 keywords × 2 pages × 10 credits = 200 credits a day, or 6,000 a month. Even if a fifth of your list escalated every single day, that's 120,000 credits a month, inside API Starter. So the working budget for 1,000 keywords is roughly $75 + $49 = $124 a month, and a freelancer with 300 keywords pays the same $75 for proxies with a fraction of the credit use.

Two things change the maths. On the flat-rate gateway, a retry costs time, never money, which is why the ladder spends its retries there first. On the API, the docs refund credits for failed jobs (500 and 530), but they don't promise that a challenge page delivered with a 200 counts as a failure, so budget as if it's billed and keep the classifier in front of every escalation. Track the real numbers from the credits column rather than trusting this table: SELECT date_trunc('month', checked_on), SUM(credits) FROM rank_checks GROUP BY 1.

Free data is worth wiring in too. The Search Console Search Analytics API reports a position field documented as "Average position in search results" for your own verified properties. It's an average across impressions, not a point-in-time check from one location, and it can't see competitors, so it complements the tracker rather than replacing it. When the two disagree by a lot for weeks, look at your location settings first.