Python-第24篇《网络编程：获取实时财经数据》

导语

今天咱们造一个“汇率雷达”——自动调用免费API获取实时汇率，JSON数据一键解析，监控到理想点位自动提醒。学会这招，你还能爬金价、股价、原油价格！

---

本篇目标

• 理解API和JSON数据格式
• 掌握requests库调用RESTful API
• 学会解析和提取JSON数据
• 实现带提醒功能的汇率监控器
• 数据本地化保存和历史追踪

---

一、准备工作：安装库

# 命令行运行
# pip install requests pandas

import requests
import json
import time
from datetime import datetime
import pandas as pd
import os
from typing import Dict, List

print("✅ 网络编程环境准备就绪！")

---

二、什么是API和JSON？

2.1 API = 数据“外卖服务”

API（应用程序接口）就像餐厅的外卖服务：

• 你下单（发送请求URL）
• 厨房做饭（服务器处理）
• 外卖小哥送上门（返回JSON数据）

免费财经API推荐：

• 汇率：https://api.exchangerate-api.com/v4/latest/USD（免费，无需注册）
• 国内汇率：https://www.banhou.com/api/v1/exchange_rate（模拟）
• 黄金价格：https://api.gold-api.com/price/XAU（模拟）

---

2.2 JSON = 数据“快递盒”

JSON是API返回的数据格式，长得像Python字典：

# 示例汇率API返回的JSON
{
  "result": "success",
  "base_code": "USD",
  "time_last_update_utc": "2025-01-26 12:30:02",
  "rates": {
    "CNY": 7.2456,
    "EUR": 0.9234,
    "GBP": 0.7856,
    "JPY": 149.85
  }
}

---

三、第一次API调用：获取实时汇率

# ========== 基础API调用 ==========
def get_exchange_rate(base_currency: str = 'USD', target_currency: str = 'CNY') -> Dict:
    """
    获取实时汇率
    base_currency: 基础货币（如USD）
    target_currency: 目标货币（如CNY）
    返回：{'汇率': float, '更新时间': str, '状态': str}
    """
    # API URL（免费版，每小时更新）
    url = f"https://api.exchangerate-api.com/v4/latest/{base_currency}"
    
    try:
        print(f"\n🌐 正在请求: {url}")
        
        # 发送GET请求（timeout防止卡死）
        response = requests.get(url, timeout=10)
        
        # 检查状态码（200表示成功）
        if response.status_code != 200:
            return {
                '状态': f'失败',
                '错误': f'Status Code: {response.status_code}',
                '汇率': 0
            }
        
        # 解析JSON
        data = response.json()
        
        # 提取目标汇率
        if 'rates' in data and target_currency in data['rates']:
            rate = data['rates'][target_currency]
            
            return {
                '汇率': rate,
                '更新时间': data.get('time_last_update_utc', '未知'),
                '状态': '成功',
                '基础货币': data.get('base_code', base_currency),
                '目标货币': target_currency
            }
        else:
            return {
                '状态': '失败',
                '错误': f'未找到货币 {target_currency}',
                '汇率': 0
            }
    
    except requests.exceptions.Timeout:
        return {'状态': '失败', '错误': '请求超时（10秒）', '汇率': 0}
    
    except requests.exceptions.ConnectionError:
        return {'状态': '失败', '错误': '网络连接错误', '汇率': 0}
    
    except Exception as e:
        return {'状态': '失败', '错误': str(e), '汇率': 0}

# 测试
print("汇率查询测试：")
result = get_exchange_rate('USD', 'CNY')
if result['状态'] == '成功':
    print(f"✅ 1美元 = {result['汇率']:.4f}人民币")
    print(f"⏰ 更新时间: {result['更新时间']}")
else:
    print(f"❌ 查询失败: {result['错误']}")

# 查询其他货币
result_eur = get_exchange_rate('USD', 'EUR')
print(f"💶 1美元 = {result_eur['汇率']:.4f}欧元")

---

四、多货币汇率监控器

# ========== 多货币监控器 ==========
class ExchangeRateMonitor:
    """汇率监控器"""
    
    def __init__(self, base_currency: str = 'USD'):
        self.base_currency = base_currency
        self.target_currencies = ['CNY', 'EUR', 'GBP', 'JPY']
        self.history_file = f"汇率历史_{base_currency}.json"
        self.load_history()
    
    def load_history(self):
        """加载历史汇率"""
        if os.path.exists(self.history_file):
            try:
                with open(self.history_file, 'r', encoding='utf-8') as f:
                    self.history = json.load(f)
            except:
                self.history = {}
        else:
            self.history = {}
    
    def save_history(self):
        """保存历史汇率"""
        with open(self.history_file, 'w', encoding='utf-8') as f:
            json.dump(self.history, f, ensure_ascii=False, indent=2)
    
    def get_all_rates(self) -> Dict:
        """获取所有目标货币汇率"""
        print(f"\n{'='*50}")
        print(f"批量查询 {self.base_currency} 汇率")
        print(f"{'='*50}")
        
        rates = {}
        
        for currency in self.target_currencies:
            result = get_exchange_rate(self.base_currency, currency)
            
            if result['状态'] == '成功':
                rates[currency] = result['汇率']
                print(f"✅ {self.base_currency}/{currency}: {result['汇率']:.4f}")
            else:
                rates[currency] = 0
                print(f"❌ {self.base_currency}/{currency}: {result['错误']}")
            
            # 礼貌性暂停1秒（避免频繁请求）
            time.sleep(1)
        
        # 记录到历史
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        self.history[timestamp] = rates
        self.save_history()
        
        print(f"\n💾 已保存到历史记录")
        
        return rates
    
    def show_history(self, n: int = 10):
        """显示最近n条历史记录"""
        print(f"\n{'='*50}")
        print(f"最近 {n} 次汇率查询历史")
        print(f"{'='*50}")
        
        # 按时间排序，取最近的
        sorted_history = sorted(self.history.items(), key=lambda x: x[0], reverse=True)
        
        for timestamp, rates in sorted_history[:n]:
            print(f"\n⏰ {timestamp}")
            for currency, rate in rates.items():
                if rate > 0:
                    print(f"  {self.base_currency}/{currency}: {rate:.4f}")
    
    def monitor_rate(self, target_currency: str, target_rate: float, direction: str = 'above'):
        """
        监控汇率达到目标值
        target_currency: 目标货币
        target_rate: 目标汇率
        direction: 'above'（高于）或 'below'（低于）
        """
        print(f"\n{'='*60}")
        print(f"启动汇率监控: {self.base_currency}/{target_currency}")
        print(f"目标: 当汇率 {'高于' if direction == 'above' else '低于'} {target_rate:.4f} 时提醒")
        print(f"{'='*60}")
        
        check_count = 0
        max_checks = 20  # 最多检查20次
        
        try:
            while check_count < max_checks:
                check_count += 1
                now = datetime.now().strftime('%H:%M:%S')
                print(f"\n第 {check_count}/{max_checks} 次检查 - {now}")
                
                result = get_exchange_rate(self.base_currency, target_currency)
                
                if result['状态'] == '成功':
                    current_rate = result['汇率']
                    print(f"当前汇率: {current_rate:.4f}")
                    
                    # 检查是否达到目标
                    if direction == 'above' and current_rate >= target_rate:
                        print(f"🎉 达到目标！汇率已高于 {target_rate:.4f}")
                        self._send_alert(self.base_currency, target_currency, current_rate, target_rate)
                        break
                    
                    elif direction == 'below' and current_rate <= target_rate:
                        print(f"🎉 达到目标！汇率已低于 {target_rate:.4f}")
                        self._send_alert(self.base_currency, target_currency, current_rate, target_rate)
                        break
                    
                    else:
                        print(f"未达到目标，继续监控...")
                
                else:
                    print(f"查询失败: {result['错误']}")
                
                # 每30秒检查一次（避免频繁请求）
                time.sleep(30)
            
            if check_count >= max_checks:
                print(f"\n⏰ 监控结束，已达到最大检查次数 {max_checks}")
        
        except KeyboardInterrupt:
            print("\n\n用户手动停止监控")
    
    def _send_alert(self, base: str, target: str, current: float, target_rate: float):
        """发送提醒（这里用打印模拟，可扩展为邮件/微信）"""
        print("\n" + "="*50)
        print("🚨 汇率提醒")
        print("="*50)
        print(f"货币对: {base}/{target}")
        print(f"当前汇率: {current:.4f}")
        print(f"目标汇率: {target_rate:.4f}")
        print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("="*50)
        print("\a")  # 发出提示音（Windows）

# 测试监控器
monitor = ExchangeRateMonitor('USD')

# 查询所有汇率
rates = monitor.get_all_rates()

# 查看历史
monitor.show_history(5)

# 启动监控（示例：监控USD/CNY是否跌破7.20）
# monitor.monitor_rate('CNY', 7.20, direction='below')

---

五、实战：实时汇率监控器（完整版）

# ========== 实时汇率监控器 - 完整版 ==========
import requests
import json
import time
from datetime import datetime
import os
from typing import Dict, List
import pandas as pd

class RealtimeExchangeMonitor:
    """实时汇率监控器（专业版）"""
    
    def __init__(self, config_file='monitor_config.json'):
        self.config_file = config_file
        self.monitoring = False
        self.load_config()
    
    def load_config(self):
        """加载监控配置"""
        default_config = {
            "base_currency": "USD",
            "monitor_pairs": [
                {"currency": "CNY", "target": 7.25, "direction": "below", "active": True},
                {"currency": "EUR", "target": 0.92, "direction": "above", "active": False},
                {"currency": "JPY", "target": 150, "direction": "above", "active": False}
            ],
            "check_interval": 300,  # 检查间隔（秒），5分钟
            "alert_sound": True,
            "save_history": True,
            "history_file": "exchange_history.csv"
        }
        
        if os.path.exists(self.config_file):
            try:
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    self.config = json.load(f)
                print("✅ 配置加载成功")
            except:
                self.config = default_config
                print("⚠️ 配置加载失败，使用默认配置")
        else:
            self.config = default_config
            self.save_config()
            print("📄 创建默认配置文件")
    
    def save_config(self):
        """保存配置"""
        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(self.config, f, ensure_ascii=False, indent=2)
    
    def add_monitor_pair(self):
        """添加监控货币对"""
        print("\n" + "="*40)
        print("添加监控货币对")
        print("="*40)
        
        base = input("基础货币（如USD）: ").strip().upper()
        target = input("目标货币（如CNY）: ").strip().upper()
        target_rate = float(input("目标汇率: ").strip())
        direction = input("方向（above/below）: ").strip().lower()
        
        new_pair = {
            "currency": target,
            "target": target_rate,
            "direction": direction,
            "active": True
        }
        
        self.config["monitor_pairs"].append(new_pair)
        self.save_config()
        
        print(f"✅ 已添加 {base}/{target} 监控")
    
    def get_rate_with_fallback(self, base: str, target: str) -> Dict:
        """获取汇率（带备用API）"""
        # 主API
        try:
            result = get_exchange_rate(base, target)
            if result['状态'] == '成功':
                return result
        except:
            pass
        
        # 备用API（这里用另一个免费API）
        try:
            print("尝试备用API...")
            url = f"https://api.frankfurter.app/latest?base={base}&symbols={target}"
            response = requests.get(url, timeout=5)
            data = response.json()
            
            if 'rates' in data and target in data['rates']:
                return {
                    '汇率': data['rates'][target],
                    '更新时间': data.get('date', ''),
                    '状态': '成功',
                    '基础货币': base,
                    '目标货币': target
                }
        except Exception as e:
            print(f"备用API也失败: {e}")
        
        return {'状态': '失败', '错误': '所有API不可用', '汇率': 0}
    
    def save_to_history(self, rates: Dict):
        """保存到历史记录CSV"""
        if not self.config['save_history']:
            return
        
        timestamp = datetime.now()
        history_file = self.config['history_file']
        
        # 准备数据
        row = {
            '时间戳': timestamp.strftime('%Y-%m-%d %H:%M:%S'),
            '基础货币': self.config['base_currency']
        }
        
        # 添加各货币汇率
        for currency, rate in rates.items():
            row[f'{currency}_汇率'] = rate
        
        # 读取或创建DataFrame
        if os.path.exists(history_file):
            df = pd.read_csv(history_file)
            df = pd.concat([df, pd.DataFrame([row])], ignore_index=True)
        else:
            df = pd.DataFrame([row])
        
        # 保存（只保留最近1000条）
        df = df.tail(1000)
        df.to_csv(history_file, index=False)
        
        print(f"💾 历史记录已更新: {history_file}")
    
    def show_history_chart(self):
        """显示汇率走势图（需要matplotlib）"""
        try:
            import matplotlib.pyplot as plt
            
            history_file = self.config['history_file']
            
            if not os.path.exists(history_file):
                print("⚠️ 暂无历史数据")
                return
            
            df = pd.read_csv(history_file)
            df['时间戳'] = pd.to_datetime(df['时间戳'])
            
            # 绘制各货币走势
            plt.figure(figsize=(12, 6))
            
            currencies = ['CNY', 'EUR', 'JPY', 'GBP']
            colors = ['red', 'blue', 'green', 'orange']
            
            for i, currency in enumerate(currencies):
                col = f'{currency}_汇率'
                if col in df.columns:
                    plt.plot(df['时间戳'], df[col], marker='o', label=f'{self.config["base_currency"]}/{currency}', color=colors[i])
            
            plt.title(f'{self.config["base_currency"]}汇率走势', fontsize=14)
            plt.xlabel('时间')
            plt.ylabel('汇率')
            plt.legend()
            plt.xticks(rotation=45)
            plt.grid(True, linestyle='--', alpha=0.6)
            plt.tight_layout()
            
            plt.savefig(f'汇率走势图_{datetime.now().strftime("%Y%m%d")}.png', dpi=150)
            plt.show()
            
        except ImportError:
            print("⚠️ 未安装matplotlib，无法显示图表")
    
    def run_monitor(self):
        """运行监控"""
        print("\n🤖 启动汇率自动监控系统")
        print(f"监控基础货币: {self.config['base_currency']}")
        print(f"检查间隔: {self.config['check_interval']}秒")
        
        active_pairs = [p for p in self.config['monitor_pairs'] if p['active']]
        
        if not active_pairs:
            print("⚠️ 没有激活的监控项")
            return
        
        self.monitoring = True
        
        try:
            check_count = 0
            
            while self.monitoring:
                check_count += 1
                print(f"\n{'='*50}")
                print(f"第 {check_count} 次检查 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
                print(f"{'='*50}")
                
                # 获取所有激活的货币汇率
                rates = {}
                for pair in active_pairs:
                    target = pair['currency']
                    result = self.get_rate_with_fallback(self.config['base_currency'], target)
                    
                    if result['状态'] == '成功':
                        rate = result['汇率']
                        rates[target] = rate
                        
                        # 检查是否达到目标
                        target_rate = pair['target']
                        direction = pair['direction']
                        
                        if direction == 'above' and rate >= target_rate:
                            print(f"🎯 {self.config['base_currency']}/{target} 达到目标！")
                            self._send_alert(self.config['base_currency'], target, rate, target_rate)
                        elif direction == 'below' and rate <= target_rate:
                            print(f"🎯 {self.config['base_currency']}/{target} 达到目标！")
                            self._send_alert(self.config['base_currency'], target, rate, target_rate)
                        else:
                            diff = abs(rate - target_rate)
                            print(f"  {target}: {rate:.4f} (距离目标{target_rate:.4f}差{diff:.4f})")
                    else:
                        print(f"  {target}: 查询失败")
                    
                    time.sleep(1)  # 礼貌延迟
                
                # 保存历史
                if rates:
                    self.save_to_history(rates)
                
                # 等待下次检查
                print(f"\n⏰ 等待 {self.config['check_interval']} 秒后下次检查...")
                time.sleep(self.config['check_interval'])
        
        except KeyboardInterrupt:
            print("\n\n用户手动停止监控")
            self.monitoring = False
    
    def _send_alert(self, base: str, target: str, current: float, target_rate: float):
        """发送提醒"""
        print("\n" + "="*50)
        print("🚨 汇率监控提醒")
        print("="*50)
        print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"货币对: {base}/{target}")
        print(f"当前汇率: {current:.4f}")
        print(f"目标汇率: {target_rate:.4f}")
        
        if current >= target_rate:
            print("📈 汇率上涨突破目标！")
        else:
            print("📉 汇率下跌跌破目标！")
        
        # 发出提示音
        if self.config.get('alert_sound', False):
            print('\a')  # 响铃
        
        print("="*50)
        
        # 可选：保存提醒记录
        alert_file = "汇率提醒记录.txt"
        with open(alert_file, 'a', encoding='utf-8') as f:
            f.write(f"{datetime.now()} - {base}/{target}: {current:.4f} (目标: {target_rate:.4f})\n")

def main():
    """主菜单"""
    monitor = RealtimeExchangeMonitor()
    
    print("=" * 60)
    print("实时汇率监控器（专业版）")
    print("支持多货币对监控，自动记录历史，触发提醒")
    print("=" * 60)
    
    while True:
        print("\n" + "=" * 45)
        print("功能菜单")
        print("=" * 45)
        print("1. 查询当前汇率")
        print("2. 查看历史记录")
        print("3. 添加监控货币对")
        print("4. 查看配置")
        print("5. 启动自动监控")
        print("6. 显示汇率走势图")
        print("7. 退出")
        print("=" * 45)
        
        choice = input("请选择: ").strip()
        
        if choice == "1":
            base = input("基础货币（默认USD）: ").strip().upper() or 'USD'
            target = input("目标货币（如CNY）: ").strip().upper()
            
            if target:
                result = monitor.get_rate_with_fallback(base, target)
                if result['状态'] == '成功':
                    print(f"\n✅ 1 {base} = {result['汇率']:.4f} {target}")
                    print(f"⏰ 更新时间: {result['更新时间']}")
        
        elif choice == "2":
            monitor.show_history(10)
        
        elif choice == "3":
            monitor.add_monitor_pair()
        
        elif choice == "4":
            print("\n当前配置:")
            print(json.dumps(monitor.config, ensure_ascii=False, indent=2))
        
        elif choice == "5":
            print("\n启动前请确保配置正确。")
            confirm = input("确认启动自动监控？(y/n): ").strip()
            if confirm.lower() == 'y':
                monitor.run_monitor()
        
        elif choice == "6":
            monitor.show_history_chart()
        
        elif choice == "7":
            print("👋 感谢使用，监控已停止！")
            monitor.monitoring = False
            break
        
        else:
            print("请输入1-7！")

if __name__ == "__main__":
    main()

---

六、知识点加油站

6.1 JSON解析技巧

"""
JSON结构：
{
  "key1": "value1",
  "key2": 123,
  "key3": {
    "nested": "value"
  },
  "key4": ["item1", "item2"]
}

Python解析：
data = response.json()

安全取值：
name = data.get('name', '默认值')  # key不存在不报错
# 等同于：
name = data['name'] if 'name' in data else '默认值'

嵌套取值（容易出错）：
# 不推荐
value = data['a']['b']['c']  # 如果a或b不存在会报错

# 推荐
value = data.get('a', {}).get('b', {}).get('c', 默认值)

类型转换：
# JSON数字 → Python
rate = float(data['rate'])

# JSON字符串 → Python
name = str(data['name'])

# JSON数组 → Python列表
items = list(data['items'])
"""

---

6.2 API调用最佳实践

"""
1. 超时设置
   requests.get(url, timeout=10)  # 10秒没响应就放弃

2. 错误重试
   for i in range(3):  # 重试3次
       try:
           response = requests.get(url, timeout=5)
           break
       except:
           time.sleep(2)
           continue

3. 请求头
   headers = {
       'User-Agent': 'MyApp/1.0',
       'Accept': 'application/json'
   }
   requests.get(url, headers=headers)

4. 参数传递
   params = {'base': 'USD', 'symbols': 'CNY,EUR'}
   requests.get(url, params=params)  # 自动转 ?base=USD&symbols=CNY,EUR

5. 响应检查
   if response.status_code == 200:  # 成功
       data = response.json()
   elif response.status_code == 404:  # 未找到
       print("API地址错误")
   elif response.status_code == 429:  # 请求过多
       print("API限流，请稍后再试")
"""

---

七、总结

• ✅ API和JSON的基础概念
• ✅ 用requests库调用RESTful API
• ✅ JSON数据的解析和提取
• ✅ 多货币汇率监控器
• ✅ 历史数据保存和可视化

---

下篇预告

第25篇《总结与展望：Python学习路线图》——回顾25篇学习旅程，规划进阶路线，推荐学习资源和实战项目！

---

🤖 Powered by Kimi K2 Thinking 💻 内容经葵葵🌻审核与修改

---