SOP: SERP Rank Tracking Fresh

Use task-based SERP endpoints for rank tracking — not Live endpoints. Task-based costs significantly less and is built for batch operations.

Process Overview

flowchart TD
    A[Build keyword list] --> B[POST tasks to /serp/google/organic/tasks_post]
    B --> C[Store task IDs]
    C --> D{Poll /tasks_ready}
    D -->|No tasks ready| E[Wait 30-60 seconds]
    E --> D
    D -->|Tasks ready| F[GET /tasks_get for each task ID]
    F --> G[Parse position data]
    G --> H[Store results with timestamp]
    H --> I[Next tracking cycle]
    I --> A

Step 1: Post Keyword Tasks

Send up to 100 keywords per request. Each keyword becomes one task.

import os
import requests
from requests.auth import HTTPBasicAuth
import json

auth = HTTPBasicAuth(
    os.environ["DATAFORSEO_LOGIN"],
    os.environ["DATAFORSEO_PASSWORD"]
)

keywords = [
    "roofing contractor miami",
    "roof repair miami fl",
    "metal roofing miami",
    "emergency roof repair miami",
    "commercial roofing miami"
]

# Build task payload — one dict per keyword
tasks = []
for keyword in keywords:
    tasks.append({
        "keyword": keyword,
        "location_code": 1015116,  # Miami, FL
        "language_code": "en",
        "device": "desktop",
        "os": "windows",
        "depth": 100  # fetch top 100 results
    })

response = requests.post(
    "https://api.dataforseo.com/v3/serp/google/organic/tasks_post",
    auth=auth,
    json=tasks
)

data = response.json()
print(f"Tasks posted: {data['tasks_count']}")
print(f"Cost: ${data['cost']}")

# Extract and store task IDs
task_ids = []
for task in data["tasks"]:
    if task["status_code"] == 20100:  # task created
        task_ids.append(task["id"])
        print(f"Task ID: {task['id']} — {task['data']['keyword']}")

# Save task IDs to a file for polling
with open("pending_task_ids.json", "w") as f:
    json.dump(task_ids, f)

Step 2: Poll for Ready Tasks

DataForSEO processes tasks asynchronously. Check /tasks_ready to know when results are available.

import time

def poll_until_ready(auth, expected_count, max_wait_minutes=10):
    """Poll tasks_ready until all tasks are done or timeout."""
    
    deadline = time.time() + (max_wait_minutes * 60)
    ready_ids = []
    
    while time.time() < deadline:
        response = requests.get(
            "https://api.dataforseo.com/v3/serp/google/organic/tasks_ready",
            auth=auth
        )
        data = response.json()
        
        if data["tasks_count"] > 0:
            for task in data["tasks"]:
                for result in task.get("result", []):
                    ready_ids.append(result["id"])
        
        print(f"Ready: {len(ready_ids)} / {expected_count}")
        
        if len(ready_ids) >= expected_count:
            break
            
        time.sleep(30)  # wait 30 seconds between polls
    
    return ready_ids

ready_task_ids = poll_until_ready(auth, expected_count=len(task_ids))

Step 3: Fetch Results

Retrieve results for each ready task ID.

def fetch_task_results(auth, task_id):
    """Fetch and return results for a single task."""
    
    response = requests.get(
        f"https://api.dataforseo.com/v3/serp/google/organic/task_get/advanced/{task_id}",
        auth=auth
    )
    return response.json()

all_results = []
for task_id in ready_task_ids:
    data = fetch_task_results(auth, task_id)
    
    if data["tasks"]:
        task = data["tasks"][0]
        if task.get("result"):
            all_results.append({
                "task_id": task_id,
                "keyword": task["data"]["keyword"],
                "result": task["result"][0]
            })

Step 4: Parse Position Data

Extract rank position and key metadata for each keyword.

from datetime import datetime

def parse_position(result_data):
    """Extract position, URL, and title for a target domain."""
    
    target_domain = "example.com"  # change to your client's domain
    items = result_data.get("items", [])
    
    for item in items:
        if item.get("type") != "organic":
            continue
        
        url = item.get("url", "")
        if target_domain in url:
            return {
                "position": item["rank_absolute"],
                "url": url,
                "title": item.get("title", ""),
                "description": item.get("description", ""),
                "found": True
            }
    
    return {"position": None, "found": False}

# Parse all results
timestamp = datetime.utcnow().isoformat()
tracking_records = []

for entry in all_results:
    position_data = parse_position(entry["result"])
    
    record = {
        "timestamp": timestamp,
        "keyword": entry["keyword"],
        "position": position_data["position"],
        "url": position_data.get("url"),
        "title": position_data.get("title"),
        "found": position_data["found"],
        "total_results": entry["result"].get("se_results_count")
    }
    
    tracking_records.append(record)
    status = f"#{record['position']}" if record["found"] else "not ranked"
    print(f"{record['keyword']}: {status}")

Step 5: Store Historical Data

Append to a CSV or database for trend tracking.

import csv

def append_to_csv(records, filepath="rank_history.csv"):
    """Append tracking records to a CSV file."""
    
    fieldnames = ["timestamp", "keyword", "position", "url", "title", "found", "total_results"]
    
    # Check if file exists to decide whether to write header
    import os
    write_header = not os.path.exists(filepath)
    
    with open(filepath, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if write_header:
            writer.writeheader()
        writer.writerows(records)
    
    print(f"Saved {len(records)} records to {filepath}")

append_to_csv(tracking_records)

Location Codes Reference

Location	Code
United States	2840
Miami, FL	1015116
New York, NY	1023191
Los Angeles, CA	1013962
Chicago, IL	1016367
Phoenix, AZ	1023080
United Kingdom	2826
Canada	2124
Australia	2036

Use /v3/serp/google/locations to get the full list of available locations.

Key Notes

• Task-based endpoints are ~10x cheaper than Live endpoints for bulk work
• Results are typically ready within 30-120 seconds
• Tasks expire after 30 days — fetch results before then
• depth: 100 fetches the top 100 positions; use depth: 10 for spot checks to save cost
• Always track by domain match, not exact URL — URLs can change