Sub GetAPIDataWithVBA()    Dim httpRequest As Object    Dim url As String    Dim responseText As String    ' 创建HTTP请求对象    Set httpRequest = CreateObject("MSXML2.XMLHTTP")    url = "https://api.openweathermap.org/data/2.5/weather?q=Beijing&appid=your_api_key"    ' 发送GET请求    httpRequest.Open "GET", url, False    httpRequest.send    ' 检查请求状态    If httpRequest.Status = 200 Then        responseText = httpRequest.responseText        ' 处理返回的JSON数据        Call ProcessJSONData(responseText)    Else        MsgBox "API请求失败，状态码：" & httpRequest.Status    End IfEnd Sub

Function ParseJSONWithVBA(jsonString As String) As Object    On Error GoTo ErrorHandler    Dim sc As Object    Set sc = CreateObject("ScriptControl")    sc.Language = "JScript"    ' 注入JSON解析函数    sc.AddCode "function parseJson(json){ return JSON.parse(json); }"    Set ParseJSONWithVBA = sc.Run("parseJson", jsonString)    Exit FunctionErrorHandler:    Set ParseJSONWithVBA = Nothing    Err.Raise 1001, "JSON Parser", "解析失败: " & Err.DescriptionEnd FunctionSub ProcessJSONData(jsonString As String)    Dim jsonObject As Object    Set jsonObject = ParseJSONWithVBA(jsonString)    ' 访问解析后的数据    Dim temperature As Double    temperature = jsonObject("main")("temp") - 273.15 ' 开尔文转摄氏度    ' 输出到Excel单元格    ThisWorkbook.Sheets("数据").Range("A1").Value = "当前温度"    ThisWorkbook.Sheets("数据").Range("B1").Value = temperature & "℃"End Sub

Sub RobustAPICall()    On Error GoTo ErrorHandler    Dim retryCount As Integer    Const maxRetries As Integer = 3    For retryCount = 1 To maxRetries        Try            Call GetAPIDataWithVBA            Exit Sub        Catch ex As Exception            If retryCount = maxRetries Then                Err.Raise 1002, "API调用", "多次重试后仍失败"            End If            ' 等待后重试            Application.Wait Now + TimeValue("00:00:02")        End Try    Next retryCountErrorHandler:    Call LogError("API调用失败: " & Err.Description)End Sub

import requestsimport jsondef get_weather_data(city_name, api_key):    """获取天气数据并解析"""    url = f"https://api.openweathermap.org/data/2.5/weather?q={city_name}&appid={api_key}"    try:        response = requests.get(url)        response.raise_for_status()  # 自动处理HTTP错误        # 直接解析JSON响应        data = response.json()        # 提取所需信息        temperature_kelvin = data['main']['temp']        temperature_celsius = temperature_kelvin - 273.15        weather_description = data['weather'][0]['description']        return {            'temperature': round(temperature_celsius, 1),            'description': weather_description,            'city': data['name']        }    except requests.exceptions.RequestException as e:        print(f"API请求失败: {e}")        return None# 使用示例weather_info = get_weather_data("Beijing", "your_api_key")if weather_info:    print(f"城市: {weather_info['city']}")    print(f"温度: {weather_info['temperature']}℃")    print(f"天气: {weather_info['description']}")

def get_authenticated_api_data(api_url, api_key):    """处理需要认证的API请求"""    headers = {        'Authorization': f'Bearer {api_key}',        'Content-Type': 'application/json'    }    try:        response = requests.get(api_url, headers=headers, timeout=10)        response.raise_for_status()        return response.json()    except requests.exceptions.Timeout:        print("请求超时")        return None    except requests.exceptions.HTTPError as err:        print(f"HTTP错误: {err}")        return None# 分页获取所有数据def get_paginated_api_data(base_url, api_key):    """处理分页API数据"""    all_data = []    page = 1    while True:        url = f"{base_url}?page={page}&limit=100"        data = get_authenticated_api_data(url, api_key)        if not data or len(data) == 0:            break        all_data.extend(data)        page += 1        # 避免过于频繁的请求        time.sleep(0.5)    return all_data

import pandas as pdfrom datetime import datetime, timedeltaclass APIDataManager:    """API数据管理器，支持缓存和批量处理"""    def __init__(self, cache_duration=3600):  # 默认缓存1小时        self.cache_duration = cache_duration        self.cache = {}    def get_cached_data(self, cache_key):        """获取缓存数据"""        if cache_key in self.cache:            cached_time, data = self.cache[cache_key]            if datetime.now() - cached_time < timedelta(seconds=self.cache_duration):                return data        return None    def get_multiple_cities_weather(self, cities, api_key):        """批量获取多个城市天气数据"""        results = []        for city in cities:            cache_key = f"weather_{city}"            cached_data = self.get_cached_data(cache_key)            if cached_data:                results.append(cached_data)            else:                data = get_weather_data(city, api_key)                if data:                    self.cache[cache_key] = (datetime.now(), data)                    results.append(data)                # 避免触发API限流                time.sleep(0.2)        return pd.DataFrame(results)# 使用示例manager = APIDataManager()cities = ["Beijing", "Shanghai", "Guangzhou", "Shenzhen"]weather_df = manager.get_multiple_cities_weather(cities, "your_api_key")print(weather_df)

from tenacity import retry, stop_after_attempt, wait_exponentialclass RobustAPIClient:    """健壮的API客户端，带重试机制"""    def __init__(self, base_retry_delay=1, max_retries=3):        self.base_retry_delay = base_retry_delay        self.max_retries = max_retries    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))    def reliable_api_call(self, url, headers=None, timeout=30):        """带重试的API调用"""        try:            response = requests.get(url, headers=headers, timeout=timeout)            response.raise_for_status()            return response.json()        except requests.exceptions.ConnectionError:            print("网络连接错误，重试中...")            raise        except requests.exceptions.Timeout:            print("请求超时，重试中...")            raise    def get_with_fallback(self, primary_url, fallback_url, headers=None):        """带降级方案的API调用"""        try:            return self.reliable_api_call(primary_url, headers)        except Exception as e:            print(f"主API失败: {e}，尝试备用API")            try:                return self.reliable_api_call(fallback_url, headers)            except Exception as fallback_error:                print(f"备用API也失败: {fallback_error}")                return None

import requestsimport pandas as pdimport timeimport smtplibfrom datetime import datetimefrom email.mime.text import MIMETextclass StockMonitor:    """股票价格监控系统"""    def __init__(self, api_key, alert_emails):        self.api_key = api_key        self.alert_emails = alert_emails        self.stock_data = pd.DataFrame()    def fetch_stock_prices(self, symbols):        """获取多支股票价格"""        base_url = "https://api.example-stock-api.com/v1/quote"        results = []        for symbol in symbols:            url = f"{base_url}?symbol={symbol}&apikey={self.api_key}"            try:                response = requests.get(url, timeout=10)                data = response.json()                results.append({                    'symbol': symbol,                    'price': data['currentPrice'],                    'change': data['change'],                    'percent_change': data['changePercent'],                    'timestamp': datetime.now()                })            except Exception as e:                print(f"获取{symbol}数据失败: {e}")            # 避免频繁请求            time.sleep(0.1)        return pd.DataFrame(results)    def check_alerts(self, alert_rules):        """检查价格预警"""        alerts = []        for _, rule in alert_rules.iterrows():            symbol = rule['symbol']            threshold = rule['threshold']            current_data = self.stock_data[                self.stock_data['symbol'] == symbol            ]            if not current_data.empty:                current_price = current_data.iloc[0]['price']                if current_price <= threshold:                    alerts.append({                        'symbol': symbol,                        'current_price': current_price,                        'threshold': threshold,                        'timestamp': datetime.now()                    })        return alerts    def send_email_alert(self, alert_info):        """发送邮件预警"""        msg = MIMEText(            f"股票{alert_info['symbol']}价格预警！\n"            f"当前价格: {alert_info['current_price']}\n"            f"预警阈值: {alert_info['threshold']}\n"            f"时间: {alert_info['timestamp']}"        )        msg['Subject'] = f"股票预警 {alert_info['symbol']}"        msg['From'] = "monitor@company.com"        msg['To'] = ", ".join(self.alert_emails)        try:            smtp_server = smtplib.SMTP('smtp.company.com', 587)            smtp_server.send_message(msg)            smtp_server.quit()            print(f"已发送{alert_info['symbol']}预警邮件")        except Exception as e:            print(f"邮件发送失败: {e}")    def run_monitoring(self, symbols, alert_rules, interval=300):        """运行监控循环"""        while True:            print(f"{datetime.now()} - 开始获取股票数据")            # 获取最新数据            new_data = self.fetch_stock_prices(symbols)            self.stock_data = new_data            # 检查预警            alerts = self.check_alerts(alert_rules)            for alert in alerts:                self.send_email_alert(alert)            # 保存历史数据            self.save_to_database(new_data)            print(f"本轮监控完成，发现{len(alerts)}个预警")            time.sleep(interval)# 使用示例if __name__ == "__main__":    monitor = StockMonitor(        api_key="your_stock_api_key",        alert_emails=["trader1@company.com", "trader2@company.com"]    )    symbols = ["AAPL", "GOOGL", "MSFT", "TSLA", "AMZN"]    alert_rules = pd.DataFrame([        {"symbol": "AAPL", "threshold": 150.0},        {"symbol": "GOOGL", "threshold": 2500.0}    ])    monitor.run_monitoring(symbols, alert_rules)