爱回收价格查询 API 接口的价格趋势分析应用：利用历史报价数据预测设备残值

import requestsimport hashlibimport timeimport jsonfrom datetime import datetimefrom apscheduler.schedulers.blocking import BlockingScheduler# 配置项APP_KEY = "你的App-Key"SECRET = "你的Secret"BATCH_API_URL = "https://openapi.aihuishou.com/v1/device/batch/estimate"DATA_STORAGE_PATH = "historical_prices.json"  # 本地存储（生产环境替换为数据库）# 待采集的核心设备列表（覆盖热门机型+不同配置）CORE_DEVICES = [
    {"model_id": "1001", "condition": 2, "fault": 0, "storage": "256G", "network": "国行"},
    {"model_id": "2003", "condition": 3, "fault": 0, "storage": "128G", "network": "国行"},
    {"model_id": "3005", "condition": 2, "fault": 0, "storage": "256G", "network": "国行"},
    # 可扩展至20台/批次（普通账号上限）]# 生成签名（遵循爱回收API规范）def generate_sign(timestamp):
    raw_str = APP_KEY + str(timestamp) + SECRET
    md5 = hashlib.md5(raw_str.encode("utf-8")).hexdigest()
    return md5.upper()# 单批次采集逻辑def collect_batch_prices(device_batch):
    timestamp = int(time.time() * 1000)
    headers = {
        "App-Key": APP_KEY,
        "Timestamp": str(timestamp),
        "Sign": generate_sign(timestamp),
        "Content-Type": "application/json"
    }
    try:
        resp = requests.post(
            BATCH_API_URL,
            json={"devices": device_batch},
            headers=headers,
            timeout=8
        )
        resp.raise_for_status()
        result = resp.json()
        if result["code"] == 0:
            # 组装采集数据，关联设备信息与报价
            collect_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            collected_data = []
            for dev, est in zip(device_batch, result["data"]["device_estimates"]):
                collected_data.append({
                    "collect_time": collect_time,
                    "model_id": dev["model_id"],
                    "brand": "苹果" if dev["model_id"].startswith("100") else "华为/小米",
                    "storage": dev["storage"],
                    "condition": dev["condition"],
                    "final_price": est["final_price"],
                    "valid_time": est["valid_time"]
                })
            # 写入存储（生产环境替换为数据库insert）
            with open(DATA_STORAGE_PATH, "a", encoding="utf-8") as f:
                json.dump(collected_data, f, ensure_ascii=False)
                f.write("\n")
            return True
        else:
            print(f"业务错误：{result['message']}")
            return False
    except Exception as e:
        print(f"采集失败：{str(e)}")
        return False# 定时采集（每日9:00/15:00执行，避开高峰期）def scheduled_collect():
    # 拆分设备列表（按20台/批次）
    batches = [CORE_DEVICES[i:i+20] for i in range(0, len(CORE_DEVICES), 20)]
    for batch in batches:
        success = collect_batch_prices(batch)
        if not success:
            # 失败重试（间隔3秒，最多2次）
            for _ in range(2):
                time.sleep(3)
                if collect_batch_prices(batch):
                    break
        time.sleep(0.02)  # 控制QPS≤50if __name__ == "__main__":
    scheduler = BlockingScheduler()
    scheduler.add_job(scheduled_collect, "cron", hour=["9", "15"])
    print("历史报价采集任务启动...")
    scheduler.start()

import pandas as pdfrom sklearn.linear_model import LinearRegressionfrom sklearn.metrics import r2_score# 1. 加载预处理后的历史数据df = pd.read_json("historical_prices.json", lines=True)# 筛选目标设备：iPhone 14 256G 99新target_data = df[(df["model_id"] == "1001") & (df["storage"] == "256G") & (df["condition"] == 2)]# 构建特征：上市后月份（假设2024-01为上市后第1个月）target_data["months"] = target_data["collect_time"].apply(lambda x: int(x.split("-")[1]))# 2. 训练模型X = target_data[["months"]]  # 自变量：上市后月份y = target_data["final_price"]  # 因变量：最终回收价model = LinearRegression()model.fit(X, y)# 3. 验证模型精度y_pred = model.predict(X)print(f"模型拟合度R²：{r2_score(y, y_pred):.2f}")  # R²≥0.8表示拟合效果好# 4. 预测未来3个月价格future_months = pd.DataFrame({"months": [10, 11, 12]})future_prices = model.predict(future_months)for month, price in zip(future_months["months"], future_prices):
    print(f"上市后第{month}个月，预测残值：{round(price, 2)}元")