招投标系统怎么用 IP 识别围标串标？投标文件制作 IP 的聚合预警（Python 实现）

公共资源交易平台的运营方和做电子招投标 SaaS 的团队，几乎都会被同一个问题找上门：「这几家投标人的文件制作 IP 一模一样，是不是串标？」

这个问题不好答。IP 相同确实是最常见的一条线索，但只把 IP 拿出来做字符串比对，往往会掉进两个坑：一是把「同一家代理公司给三家投标人做标书」这种合规情形判成串标；二是三家公司用了同一个云桌面、同一个网吧、或者同一个地市运营商的出口，也会被判成串标，投诉和质疑很快就来了。法院的判例态度也很明确——IP 一致是线索，不是定性结论。

所以真正能用的做法是：先把 IP 做属地化（拿到省市、运营商），再在项目维度上做聚合分析，输出「有理由的怀疑」和可复核的证据描述，最后交给人工判断。这篇文章给一套 Python 实现，输入是投标记录表，输出是按项目排好序的预警报告。

一、招投标场景里，哪些 IP 特征值得怀疑

先说清楚判断依据，再看代码。围标串标在 IP 层面留下的痕迹，大致有这几类，可靠度从高到低：

同一 IP 出现在多家投标人身上。 最硬的一条。不同公司用同一台机器（或者同一个出口）制作并上传标书，这本身就说明它们共享了制作环境。要注意的是，如果这个 IP 是招标代理机构所在地或者公共资源交易中心的便民服务终端，那就属于正常——这类出口 IP 要建白名单。

同一 /24 网段下的多家投标人。 同一 C 段通常意味着同一个机房、同一条专线、同一个小区宽带。它比「IP 完全相同」弱，但比「同城」强得多。云桌面厂商的出口往往就是若干个 C 段。

同一 IP 的提交时间高度接近。 三家公司在半小时内用同一个 IP 上传标书，这个组合几乎没法解释成巧合；如果是同一天但间隔好几小时，可信度就降一档。

多家投标人的制作 IP 都落在机房/云出口。 用云服务器或云桌面做标书，本身不违规（有些投标人就习惯在云上办公），但「多家投标人 + 同一城市的机房出口」这个组合值得提示核验——真实的黑产链条里，代做标书的工作室就是这么干的。

而下面这些是正常现象，规则不能一刀切：

• 投标人在异地出差，用酒店 WiFi 上传标书，IP 属地和他公司注册地完全不一致。
• 公司用运营商专线，出口 IP 显示为省会城市，实际公司在下面的地级市。
• 用 VPN 或者企业零信任网络上传，出口显示成总部城市。
• 招标代理机构统一提供标书上传服务，多家投标人的 IP 天然相同。

二、方案设计

整体就四步，没必要上复杂的架构：

投标记录（项目、投标人、制作IP、提交时间）
   → 清洗：剥端口、去空格、区内网地址
   → 属地化：SQLite 缓存 + 令牌桶限速，失败返回空而不是抛异常
   → 聚合：按项目分组，跑 4 条规则，累加风险分
   → 输出：按分数排序的预警报告（控制台 + JSON 留档）

三个设计上的取舍：

属地化必须带缓存和限速。 一个交易平台一年也就几万个投标 IP，重复率极高（代理机构、云桌面、地市运营商出口都是反复出现），缓存 30 天能把接口调用量压到原始调用的一成左右。限速用令牌桶压在 55 次/分钟，避免触发免费版 60 次/分钟的上限。

接口失败不能中断批处理。 一条 IP 查不到（格式不对、被限速、网络抖动）就让整份报告挂掉，这在生产里是不可接受的。所以解析器统一返回 None，规则层把「属地为空」当成「本次没有位置信号」，只降权不报错。

输出的是证据描述，不是结论。 每条预警都要能直接读给评标委员会或者监管部门听——所以证据里必须带 IP、属地、涉及哪几家投标人、时间差多少分钟。原始接口返回也要落库留存，后面真要走质疑/投诉流程，这些就是原始记录。

三、Python 实现

下面这份脚本只依赖标准库（连 requests 都不用装），把 SAMPLE 换成自己的 CSV 就能跑。分三个部分：限速器、属地解析器、规则引擎。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""招投标场景：投标文件制作 IP 的属地化 + 聚合预警"""
import csv
import ipaddress
import json
import sqlite3
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import defaultdict
from datetime import datetime

IP9_API = "https://ip9.com.cn/get"
GEO_TTL = 30 * 86400          # 投标 IP 复用率高，缓存 30 天
API_TIMEOUT = 3.0
RATE_PER_MIN = 55             # 免费版 60 次/分钟，留余量
IDC_HINTS = ("云", "计算", "数据中心", "IDC", "机房", "CDN")


class TokenBucket:
    def __init__(self, rate_per_min):
        self.capacity = float(rate_per_min)
        self.rate = rate_per_min / 60.0
        self.tokens = float(rate_per_min)
        self.ts = time.monotonic()
        self.lock = threading.Lock()

    def take(self):
        while True:
            with self.lock:
                now = time.monotonic()
                self.tokens = min(self.capacity, self.tokens + (now - self.ts) * self.rate)
                self.ts = now
                if self.tokens >= 1:
                    self.tokens -= 1
                    return
                wait = (1 - self.tokens) / self.rate
            time.sleep(min(wait, 0.5))


class GeoResolver:
    """IP → 属地：SQLite 缓存 + 令牌桶限速，查不到返回 None"""

    def __init__(self, db_path="tender.db"):
        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.execute("CREATE TABLE IF NOT EXISTS ip_geo (ip TEXT PRIMARY KEY, payload TEXT, ts INTEGER)")
        self.db.commit()
        self.bucket = TokenBucket(RATE_PER_MIN)
        self.api_calls = 0
        self.errors = defaultdict(int)

    def lookup(self, ip):
        row = self.db.execute("SELECT payload, ts FROM ip_geo WHERE ip=?", (ip,)).fetchone()
        if row and time.time() - row[1] < GEO_TTL:
            return json.loads(row[0])

        self.bucket.take()
        self.api_calls += 1
        url = IP9_API + "?ip=" + urllib.parse.quote(ip)
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "tender-risk/1.0"})
            with urllib.request.urlopen(req, timeout=API_TIMEOUT) as resp:
                body = json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            self.errors["http_%d" % e.code] += 1        # 400/403/429
            return None
        except (urllib.error.URLError, TimeoutError, OSError, json.JSONDecodeError):
            self.errors["network"] += 1
            return None

        if body.get("ret") != 200:
            self.errors["ret_%s" % body.get("ret")] += 1
            return None

        data = body.get("data") or {}
        geo = {
            "country": data.get("country") or "",
            "prov": data.get("prov") or "",
            "city": data.get("city") or "",
            "isp": data.get("isp") or "",
            "lat": data.get("lat") or "",
            "lng": data.get("lng") or "",
            "big_area": data.get("big_area") or "",
            "raw": data,                                   # 原始响应留证
            "queried_at": int(time.time()),
        }
        if not geo["city"]:
            self.errors["empty_location"] += 1             # 内网/保留地址、海外 IP
        self.db.execute("INSERT OR REPLACE INTO ip_geo VALUES (?,?,?)",
                        (ip, json.dumps(geo, ensure_ascii=False), int(time.time())))
        self.db.commit()
        return geo


def net24(ip):
    """取 /24 网段；IPv6 取 /48。拿不到就返回 None"""
    try:
        addr = ipaddress.ip_address(ip)
    except ValueError:
        return None
    prefix = 24 if addr.version == 4 else 48
    return str(ipaddress.ip_network("%s/%d" % (ip, prefix), strict=False))


def parse_ts(s):
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M"):
        try:
            return int(datetime.strptime(s, fmt).timestamp())
        except ValueError:
            continue
    return None


def analyze_project(project_id, rows):
    """rows: [{bidder, ip, ts, geo}]，返回该项目的预警结果"""
    score, evidence = 0, []

    # 规则 1：同一 IP 出现在多家投标人（最硬的一条）
    by_ip = defaultdict(set)
    for r in rows:
        by_ip[r["ip"]].add(r["bidder"])
    shared_ip = {ip: b for ip, b in by_ip.items() if len(b) >= 2}
    for ip, bidders in shared_ip.items():
        score += 45
        g = next(r["geo"] for r in rows if r["ip"] == ip) or {}
        where = "%s%s" % (g.get("prov", ""), g.get("city", "")) or "属地未知"
        evidence.append("IP %s（%s / %s）被 %d 家投标人共用：%s"
                        % (ip, where, g.get("isp", "-"), len(bidders), "、".join(sorted(bidders))))

    # 规则 2：同一 /24 网段（同机房出口的强特征）
    by_net = defaultdict(set)
    for r in rows:
        n = net24(r["ip"])
        if n:
            by_net[n].add(r["bidder"])
    for n, bidders in ({k: v for k, v in by_net.items() if len(v) >= 2}).items():
        if any(r["ip"] in shared_ip for r in rows if net24(r["ip"]) == n):
            continue                                    # 已被规则 1 覆盖，不重复计分
        score += 25
        evidence.append("同网段 %s 下 %d 家投标人：%s" % (n, len(bidders), "、".join(sorted(bidders))))

    # 规则 3：同一出口 IP 提交时间高度接近（30 分钟内）
    by_ip_rows = defaultdict(list)
    for r in rows:
        by_ip_rows[r["ip"]].append(r)
    for ip, rs in by_ip_rows.items():
        rs.sort(key=lambda x: x["ts"])
        for a, b in zip(rs, rs[1:]):
            if b["bidder"] != a["bidder"] and b["ts"] - a["ts"] <= 1800:
                score += 20
                evidence.append("同一 IP %s 在 %d 分钟内有 %s、%s 两家提交标书"
                                % (ip, (b["ts"] - a["ts"]) // 60, a["bidder"], b["bidder"]))

    # 规则 4：多家投标人的制作 IP 落在同一城市的机房/云出口（弱信号）
    idc_cities = defaultdict(set)
    for r in rows:
        g = r["geo"] or {}
        if any(h in (g.get("isp") or "") for h in IDC_HINTS):
            idc_cities[g.get("city", "-")].add(r["bidder"])
    for city, bidders in idc_cities.items():
        if len(bidders) >= 2:
            score += 10
            evidence.append("有 %d 家投标人的制作 IP 落在 %s 的机房/云出口（需核验是否使用代理或云桌面）"
                            % (len(bidders), city))

    return {"project_id": project_id, "score": min(score, 100),
            "level": "high" if score >= 60 else ("medium" if score >= 30 else "low"),
            "bidders": sorted({r["bidder"] for r in rows}), "evidence": evidence}

主体逻辑到这里就完整了。跑起来的方式是把投标记录读进来逐条属地化，再按项目聚合：

# 样表：真实可查的 IP，模拟「同一项目里两家用了同一个出口 IP」
SAMPLE = """project_id,bidder,doc_ip,submit_time
2026-JS-018,甲建设有限公司,114.114.114.114,2026-09-16 09:10:00
2026-JS-018,乙工程有限公司,114.114.114.114,2026-09-16 09:22:00
2026-JS-018,丙建筑公司,119.29.29.29,2026-09-16 09:31:00
2026-JS-018,丁安装公司,202.96.128.86,2026-09-16 14:02:00
2026-JS-019,戊装饰公司,8.8.8.8,2026-09-16 10:00:00
"""


def main():
    import io
    reader = csv.DictReader(io.StringIO(SAMPLE))
    resolver = GeoResolver()
    projects = defaultdict(list)
    for row in reader:
        ip = row["doc_ip"].strip().split(":")[0]        # 关键：先剥掉端口
        geo = resolver.lookup(ip)
        projects[row["project_id"]].append({
            "bidder": row["bidder"].strip(), "ip": ip,
            "ts": parse_ts(row["submit_time"].strip()) or int(time.time()), "geo": geo})
        g = geo or {}
        print("属地化 %-16s → %s%s / %s" % (ip, g.get("prov", ""), g.get("city", ""), g.get("isp", "-")))

    reports = [analyze_project(pid, rows) for pid, rows in projects.items()]
    reports.sort(key=lambda r: -r["score"])
    with open("risk_report.json", "w", encoding="utf-8") as f:
        json.dump(reports, f, ensure_ascii=False, indent=2)

    print("\n===== 预警报告 =====")
    for r in reports:
        print("[%s] %s 风险分 %d，投标人：%s" % (r["level"].upper(), r["project_id"], r["score"],
                                                "、".join(r["bidders"])))
        for e in r["evidence"]:
            print("   · " + e)
    print("\n接口调用 %d 次，异常统计：%s" % (resolver.api_calls, dict(resolver.errors)))


if __name__ == "__main__":
    main()

用一份构造的样表（同一项目里甲、乙两家用同一个 IP，间隔 12 分钟；各家 IP 都是真实可查的地址）跑一遍，实际输出是这样：

属地化 114.114.114.114  → 江苏南京 / 114DNS
属地化 114.114.114.114  → 江苏南京 / 114DNS
属地化 119.29.29.29     → 广东广州 / 腾讯云/DNSPod DNS+
属地化 202.96.128.86    → 广东广州 / 中国电信/DNS
属地化 8.8.8.8          →  / Google Cloud

===== 预警报告 =====
[HIGH] 2026-JS-018 风险分 65，投标人：丁安装公司、丙建筑公司、乙工程有限公司、甲建设公司
   · IP 114.114.114.114（江苏南京 / 114DNS）被 2 家投标人共用：乙工程有限公司、甲建设有限公司
   · 同一 IP 114.114.114.114 在 12 分钟内有 甲建设有限公司、乙工程有限公司 两家提交标书
[LOW] 2026-JS-019 风险分 0，投标人：戊装饰公司

接口调用 4 次，异常统计：{'empty_location': 1}

第二条记录用的是 8.8.8.8，返回 country=美国、isp=Google Cloud，但 prov/city 是空的——海外 IP 在国内省市字段上本来就是空的，脚本把它归到 empty_location 计数里，不参与计分，也不会让整批任务中断。这一条正好说明为什么规则层要容忍「属地为空」。

四、落地时容易踩的坑

IP 字段先洗干净再入库。 电子招投标系统里记录下来的制作 IP，经常带着端口（8.8.8.8:80）、CIDR 斜杠、前后空格，甚至有的是 X-Forwarded-For 的整串内容。带端口和非法格式的字符串查接口会直接返回 HTTP 400，白耗一次调用。代码里 row["doc_ip"].strip().split(":")[0] 这一步看着土，但能省一批无谓请求；更严谨的做法是用 ipaddress.ip_address() 先校验。

内网地址会「查得到但没有位置」。 192.168.1.1、10.0.0.1、127.0.0.1 这类地址，接口会返回 ret=200，country 是「保留」、isp 是「内网地址」，但 prov/city/经纬度全是空串。如果代码只判断 ret == 200 就当成功，后面按城市聚合时会多出一堆空城市的假分组。判断标准应该是「city 非空才算拿到位置」。

共享出口 IP 一定要设白名单。 招标代理机构、交易中心服务终端、大型企业园区出口，这三类 IP 天然会被多家投标人共用。上线前把已知的合规出口建成白名单，命中就只记录不告警，否则第一批预警就会全是误报，业务方三天之内就把规则关掉了。

海外投标或境外节点要注意字段差异。 走海外网络参与国内项目的情况这几年变多了。海外 IP 通常只有 country 和运营商，没有国内省市，big_area 也是空的。这类记录建议单独出一个「境外制作」标签，而不是塞进城市聚合里。

阈值和权重按你的项目规模调。 上面的 45/25/20/10 只是起点：投标人少的小项目（3~5 家）里「同 IP」几乎是决定性的，权重可以再高；几百家投标人的大项目里，同一 /24 网段可能只是同一栋写字楼的宽带，权重就要降。建议先只记录不告警，跑两三个项目统计各条规则的实际命中率，再决定打开哪几条。

证据要留原始记录。 前面 geo 里特意存了 raw（接口完整返回）和 queried_at（查询时间戳）。真要走质疑投诉流程时，能拿出「某年某月某日某时查询某 IP 得到的省市、运营商」这样的原始记录，比事后重新查一遍有说服力得多——IP 数据是会随时间变化的。

IP 只能当线索，写进制度里。 这一点必须写进产品说明和提示文案里：IP 一致、网段一致，都只能作为「疑似串通投标」的线索，需要结合投标文件相似度、报价规律、投标保证金来源、法定代表人关联关系等多维证据综合判断，最终由评标委员会或监管部门认定。平台的定位是「提供线索和复核工具」，不是「输出定性结论」。

合规上别越界。 采集和留存的 IP 属于个人信息，用途要在隐私政策里写明（招投标活动监管、异常行为识别），留存期限也要有说法。别顺手把 IP 归属地拿去做别的营销用途，交易平台的合规审计不是摆设。

总结

这套东西的价值不在规则多复杂，而在三件事做实了：属地化带缓存和限速（省接口额度）、失败不中断批处理（生产可用）、输出的预警带可读证据（业务方能复核）。

按部署顺序建议这么走：先把属地化管道打通、只记录不告警；再统计各条规则在真实项目上的命中率；最后逐个打开规则，同时配套建白名单和人工复核流程。

IP 属地查询用 IP9 免费版就行：https://ip9.com.cn/get?ip=<IP> 查指定 IP（IPv4/IPv6 都支持），不带参数则返回请求方自己的归属地；返回国家、省、市、邮编、区号、运营商、城市中心经纬度、long_ip、big_area（华东、华南这类大区）等字段，免注册、无需鉴权、60 次/分钟。要做区县级比对，或者需要用 ip_type 判断某个 IP 是 IDC 机房、企业专线还是家庭宽带（招投标场景里判断「是不是云桌面/代理」很有用），那就得上 VIP 版；18 万次/分钟、SLA 99.95%，接口地址和字段用法都不变。官网：https://www.ip9.com.cn

上一篇民宿、短租平台的订单怎么防异常？用 IP 归属地识别盗号下单和刷单（Java 实现）下一篇IP 归属地接口报错怎么排查？400/403/429/超时/空结果的定位思路与 Go 兜底写法