The Definitive strategies Guide for open-source and salary negotiation

import requests
import json
import os
import time
from typing import List, Dict, Optional

class GitHubContributionAnalyzer:
    """Analyzes GitHub repositories to identify high-value contribution opportunities for open-source career growth."""

    GITHUB_API_BASE = "https://api.github.com"
    RATE_LIMIT_REMAINING_HEADER = "X-RateLimit-Remaining"
    RATE_LIMIT_RESET_HEADER = "X-RateLimit-Reset"

    def __init__(self, github_token: Optional[str] = None):
        """Initialize analyzer with optional GitHub personal access token to increase rate limits.

        Args:
            github_token: Personal access token with public_repo scope. If None, uses GITHUB_TOKEN env var.
        """
        self.github_token = github_token or os.getenv("GITHUB_TOKEN")
        self.session = requests.Session()
        if self.github_token:
            self.session.headers.update({"Authorization": f"token {self.github_token}"})
        self.session.headers.update({"Accept": "application/vnd.github.v3+json"})

    def _handle_rate_limit(self, response: requests.Response) -> None:
        """Pause execution if GitHub API rate limit is exceeded, wait until reset time.

        Args:
            response: Response object from GitHub API call.
        """
        if response.status_code == 403 and "rate limit exceeded" in response.text.lower():
            reset_time = int(response.headers.get(self.RATE_LIMIT_RESET_HEADER, time.time() + 60))
            wait_seconds = reset_time - time.time()
            if wait_seconds > 0:
                print(f"Rate limit exceeded. Waiting {wait_seconds:.0f} seconds until reset.")
                time.sleep(wait_seconds + 1)  # Add 1s buffer to avoid race conditions
            raise requests.exceptions.RequestException("Rate limit exceeded, retry after waiting.")

    def get_top_repos_for_contribution(self, language: str, min_stars: int = 1000, max_repos: int = 10) -> List[Dict]:
        """Fetch top repositories for a given language with minimum star count.

        Args:
            language: Programming language to filter by (e.g., "python", "go").
            min_stars: Minimum number of stars for repos to consider.
            max_repos: Maximum number of repos to return.

        Returns:
            List of repo metadata dicts with name, stars, good_first_issues_count.
        """
        url = f"{self.GITHUB_API_BASE}/search/repositories"
        params = {
            "q": f"language:{language} stars:>{min_stars} good-first-issues:>0",
            "sort": "stars",
            "order": "desc",
            "per_page": max_repos
        }
        try:
            response = self.session.get(url, params=params)
            self._handle_rate_limit(response)
            response.raise_for_status()
            data = response.json()
            repos = []
            for item in data.get("items", []):
                # Fetch good first issues count for each repo (requires separate API call)
                issues_url = f"{self.GITHUB_API_BASE}/repos/{item['full_name']}/issues"
                issues_params = {"labels": "good first issue", "state": "open", "per_page": 1}
                issues_response = self.session.get(issues_url, params=issues_params)
                self._handle_rate_limit(issues_response)
                issues_response.raise_for_status()
                # Get total count from Link header if available, else 0
                link_header = issues_response.headers.get("Link", "")
                total_issues = 0
                if 'rel="last"' in link_header:
                    # Extract last page number from Link header to get total count
                    last_link = [link for link in link_header.split(",") if 'rel="last"' in link][0]
                    last_page = int(last_link.split("page=")[1].split(">")[0])
                    total_issues = last_page  # Assuming per_page=1, so last page = total count
                repos.append({
                    "full_name": item["full_name"],
                    "stars": item["stargazers_count"],
                    "good_first_issues": total_issues,
                    "url": item["html_url"]
                })
            return repos
        except requests.exceptions.RequestException as e:
            print(f"Error fetching repos: {e}")
            return []

if __name__ == "__main__":
    # Example usage: Analyze Python repos for contribution opportunities
    analyzer = GitHubContributionAnalyzer()  # Set GITHUB_TOKEN env var for higher rate limits
    python_repos = analyzer.get_top_repos_for_contribution(language="python", min_stars=5000, max_repos=5)
    print("Top Python Repos for Contribution:")
    for repo in python_repos:
        print(f"{repo['full_name']}: {repo['stars']} stars, {repo['good_first_issues']} good first issues")
    # Output saved to contribution_targets.json for later use
    with open("contribution_targets.json", "w") as f:
        json.dump(python_repos, f, indent=2)

import json
from typing import Dict, List, Optional
from datetime import datetime

class NegotiationScriptGenerator:
    """Generates data-backed salary negotiation scripts tailored to individual contributor value propositions."""

    def __init__(self, current_salary: float, target_salary: float, years_experience: int):
        """Initialize with core salary parameters.

        Args:
            current_salary: Current annual base salary in USD.
            target_salary: Desired annual base salary in USD.
            years_experience: Total years of professional engineering experience.

        Raises:
            ValueError: If current_salary <=0, target_salary < current_salary, or years_experience <0.
        """
        if current_salary <= 0:
            raise ValueError("Current salary must be positive.")
        if target_salary < current_salary:
            raise ValueError("Target salary must be >= current salary.")
        if years_experience < 0:
            raise ValueError("Years of experience cannot be negative.")
        self.current_salary = current_salary
        self.target_salary = target_salary
        self.years_experience = years_experience
        self.value_props: List[Dict] = []
        self.market_data: Dict = self._load_market_data()

    def _load_market_data(self) -> Dict:
        """Load localized market salary data from bundled 2024 compensation report.

        Returns:
            Dict mapping years of experience to 75th percentile salary by region.
        """
        # Bundled data from 2024 State of Developer Compensation (n=12,000)
        return {
            "us_west": {3: 145000, 5: 185000, 7: 225000, 10: 285000},
            "us_east": {3: 135000, 5: 175000, 7: 215000, 10: 265000},
            "eu_west": {3: 95000, 5: 125000, 7: 155000, 10: 195000},
            "apac": {3: 85000, 5: 115000, 7: 145000, 10: 185000}
        }

    def add_value_prop(self, description: str, impact_metric: str, quantified_value: Optional[float] = None):
        """Add a value proposition to present during negotiation.

        Args:
            description: Plain English description of the value delivered.
            impact_metric: Metric the value impacted (e.g., "p99 latency", "monthly active users").
            quantified_value: Optional numerical value of impact (e.g., 200000 for revenue saved).
        """
        self.value_props.append({
            "description": description,
            "impact_metric": impact_metric,
            "quantified_value": quantified_value,
            "added_at": datetime.now().isoformat()
        })

    def generate_script(self, region: str = "us_west") -> str:
        """Generate full negotiation script using anchoring framework v2.1.

        Args:
            region: Target region for market data adjustment.

        Returns:
            Formatted negotiation script string.
        """
        if region not in self.market_data:
            raise ValueError(f"Unsupported region: {region}. Choose from {list(self.market_data.keys())}")
        # Get 75th percentile salary for experience level
        exp_tiers = sorted(self.market_data[region].keys())
        closest_exp = min(exp_tiers, key=lambda x: abs(x - self.years_experience))
        market_75th = self.market_data[region][closest_exp]
        # Calculate ask premium over market
        market_premium = (self.target_salary - market_75th) / market_75th * 100
        script = f"""# Salary Negotiation Script (Anchoring v2.1)
Prepared: {datetime.now().strftime("%Y-%m-%d")}
Current Salary: ${self.current_salary:,.2f}
Target Salary: ${self.target_salary:,.2f}
Market 75th Percentile ({region}, {closest_exp} yrs exp): ${market_75th:,.2f}
Ask Premium Over Market: {market_premium:.1f}%

## Opening Anchor
"I’ve really enjoyed contributing to [project X] over the past 6 months, and I’m excited to take on more responsibility as [new role]. Based on my research of market rates for {self.years_experience} years of experience in {region}, and the value I’ve delivered to the team, I’m targeting a base salary of ${self.target_salary:,.0f}."

## Value Proof Section
"""
        for i, prop in enumerate(self.value_props, 1):
            prop_line = f"{i}. {prop['description']} (Improved {prop['impact_metric']}"
            if prop["quantified_value"]:
                prop_line += f" by {prop['quantified_value']:,.0f})"
            else:
                prop_line += ")"
            script += prop_line + "\n"

        script += """
## Closing Ask
"Given this track record, I’m confident that ${target_salary:,.0f} is fair for both the company and my career growth. What steps can we take to get to that number?"
        """.replace("${target_salary:,.0f}", f"${self.target_salary:,.0f}")
        return script

    def save_script(self, filepath: str = "negotiation_script.md") -> None:
        """Save generated script to file.

        Args:
            filepath: Path to save the script (default: negotiation_script.md).
        """
        script = self.generate_script()
        try:
            with open(filepath, "w") as f:
                f.write(script)
            print(f"Script saved to {filepath}")
        except IOError as e:
            print(f"Error saving script: {e}")

if __name__ == "__main__":
    # Example usage for a senior engineer with 7 years experience
    try:
        generator = NegotiationScriptGenerator(
            current_salary=175000,
            target_salary=225000,
            years_experience=7
        )
        # Add value props from recent work
        generator.add_value_prop(
            description="Led migration from monolith to microservices for payment processing",
            impact_metric="p99 payment latency",
            quantified_value=1200  # ms reduced to 200ms
        )
        generator.add_value_prop(
            description="Mentored 3 junior engineers to promotion-ready level",
            impact_metric="team retention rate",
            quantified_value=100  # 85% to 100%
        )
        generator.add_value_prop(
            description="Contributed 12 merged PRs to upstream Kubernetes repo",
            impact_metric="open-source contribution count",
            quantified_value=12
        )
        # Generate and save script for US West region
        print(generator.generate_script(region="us_west"))
        generator.save_script()
    except ValueError as e:
        print(f"Invalid input: {e}")

import sqlite3
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional

class CareerTracker:
    """Tracks open-source contributions and salary changes to measure career ROI over time."""

    def __init__(self, db_path: str = "career_tracker.db"):
        """Initialize tracker with SQLite database.

        Args:
            db_path: Path to SQLite database file.
        """
        self.db_path = db_path
        self._init_db()

    def _init_db(self) -> None:
        """Create database tables if they do not exist."""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                # Contributions table: merged PRs, repos, date, impact score
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS contributions (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        repo_full_name TEXT NOT NULL,
                        pr_number INTEGER NOT NULL,
                        merged_at TEXT NOT NULL,
                        impact_score INTEGER CHECK(impact_score BETWEEN 1 AND 10),
                        notes TEXT,
                        UNIQUE(repo_full_name, pr_number)
                    )
                """)
                # Salary changes table: date, old_salary, new_salary, reason
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS salary_changes (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        change_date TEXT NOT NULL,
                        old_salary REAL NOT NULL,
                        new_salary REAL NOT NULL,
                        reason TEXT,
                        negotiation_script_path TEXT
                    )
                """)
                # ROI summary table: period_start, period_end, contribution_count, salary_growth_pct
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS roi_summary (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        period_start TEXT NOT NULL,
                        period_end TEXT NOT NULL,
                        contribution_count INTEGER NOT NULL,
                        salary_growth_pct REAL NOT NULL
                    )
                """)
                conn.commit()
        except sqlite3.Error as e:
            print(f"Database initialization error: {e}")
            raise

    def log_contribution(self, repo_full_name: str, pr_number: int, merged_at: str, 
                         impact_score: int = 5, notes: Optional[str] = None) -> bool:
        """Log a merged open-source contribution.

        Args:
            repo_full_name: Full repo name (e.g., "kubernetes/kubernetes").
            pr_number: Merged PR number.
            merged_at: ISO format date string of merge (e.g., "2024-05-15").
            impact_score: 1-10 score of contribution impact on career (default 5).
            notes: Optional notes about the contribution.

        Returns:
            True if logged successfully, False if duplicate or error.
        """
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute("""
                    INSERT INTO contributions (repo_full_name, pr_number, merged_at, impact_score, notes)
                    VALUES (?, ?, ?, ?, ?)
                """, (repo_full_name, pr_number, merged_at, impact_score, notes))
                conn.commit()
                print(f"Logged contribution: {repo_full_name}#{pr_number}")
                return True
        except sqlite3.IntegrityError:
            print(f"Duplicate contribution: {repo_full_name}#{pr_number}")
            return False
        except sqlite3.Error as e:
            print(f"Error logging contribution: {e}")
            return False

    def log_salary_change(self, old_salary: float, new_salary: float, reason: str, 
                          change_date: Optional[str] = None, negotiation_script_path: Optional[str] = None) -> bool:
        """Log a salary change event.

        Args:
            old_salary: Previous annual salary.
            new_salary: New annual salary.
            reason: Reason for change (e.g., "negotiation", "promotion").
            change_date: ISO format date of change (default: today).
            negotiation_script_path: Path to negotiation script used (if applicable).

        Returns:
            True if logged successfully, False otherwise.
        """
        if not change_date:
            change_date = datetime.now().strftime("%Y-%m-%d")
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute("""
                    INSERT INTO salary_changes (change_date, old_salary, new_salary, reason, negotiation_script_path)
                    VALUES (?, ?, ?, ?, ?)
                """, (change_date, old_salary, new_salary, reason, negotiation_script_path))
                conn.commit()
                print(f"Logged salary change: ${old_salary:,.0f} -> ${new_salary:,.0f}")
                return True
        except sqlite3.Error as e:
            print(f"Error logging salary change: {e}")
            return False

    def generate_roi_report(self, period_months: int = 12) -> Dict:
        """Generate ROI report for a given period.

        Args:
            period_months: Number of months to include in report (default 12).

        Returns:
            Dict with contribution count, salary growth, ROI multiple.
        """
        # Calculate period start date
        period_start = (datetime.now() - timedelta(days=period_months*30)).strftime("%Y-%m-%d")
        period_end = datetime.now().strftime("%Y-%m-%d")
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                # Get contribution count in period
                cursor.execute("""
                    SELECT COUNT(*) FROM contributions 
                    WHERE merged_at >= ? AND merged_at <= ?
                """, (period_start, period_end))
                contrib_count = cursor.fetchone()[0]
                # Get latest salary change in period
                cursor.execute("""
                    SELECT old_salary, new_salary FROM salary_changes
                    WHERE change_date >= ? AND change_date <= ?
                    ORDER BY change_date DESC LIMIT 1
                """, (period_start, period_end))
                salary_row = cursor.fetchone()
                salary_growth_pct = 0.0
                if salary_row:
                    old, new = salary_row
                    salary_growth_pct = (new - old) / old * 100
                # Calculate ROI: assume $50/hour for contribution time, 4 hrs/week
                contrib_investment = contrib_count * 4 * 50  # $50/hr * 4 hrs/PR * PR count
                salary_gain = (new - old) if salary_row else 0
                roi_multiple = salary_gain / contrib_investment if contrib_investment > 0 else 0
                return {
                    "period_start": period_start,
                    "period_end": period_end,
                    "contribution_count": contrib_count,
                    "salary_growth_pct": salary_growth_pct,
                    "contrib_investment_usd": contrib_investment,
                    "salary_gain_usd": salary_gain,
                    "roi_multiple": roi_multiple
                }
        except sqlite3.Error as e:
            print(f"Error generating report: {e}")
            return {}

if __name__ == "__main__":
    # Example usage
    tracker = CareerTracker()
    # Log sample contributions
    tracker.log_contribution("kubernetes/kubernetes", 12345, "2024-03-15", impact_score=8, notes="Fixed kubelet memory leak")
    tracker.log_contribution("prometheus/prometheus", 67890, "2024-05-20", impact_score=7, notes="Added new exporter for Redis")
    # Log sample salary change
    tracker.log_salary_change(old_salary=175000, new_salary=225000, reason="negotiation", change_date="2024-06-01", negotiation_script_path="negotiation_script.md")
    # Generate 12-month ROI report
    report = tracker.generate_roi_report(period_months=12)
    print("12-Month Career ROI Report:")
    print(json.dumps(report, indent=2))

generator.add_value_prop(
    description="Reduced production incident count by 40% via improved observability",
    impact_metric="monthly SEV-1 incidents",
    quantified_value=4  # 10 incidents/month to 6
)

analyzer = GitHubContributionAnalyzer()
top_repos = analyzer.get_top_repos_for_contribution(
    language="go", 
    min_stars=10000, 
    max_repos=3
)

tracker.log_contribution(
    repo_full_name="kubernetes/kubernetes",
    pr_number=12345,
    merged_at="2024-03-15",
    impact_score=8,
    notes="Fixed kubelet memory leak"
)

oss-salary-guide/
├── LICENSE
├── README.md
├── requirements.txt
├── src/
│   ├── github_contribution_analyzer.py
│   ├── salary_negotiation_script_generator.py
│   └── career_tracker.py
├── examples/
│   ├── sample_contribution_targets.json
│   ├── sample_negotiation_script.md
│   └── sample_career_tracker.db
└── tests/
    ├── test_github_analyzer.py
    ├── test_negotiation_generator.py
    └── test_career_tracker.py