系列文章目录
Python数据科学家养成计划（Python学习指南）
前言
随着人工智能的不断发展，数据科学相关技术也越来越重要，很多人都开启了学习数据科学相关知识体系的学习之旅，本文就以 《Python CookBook》一书为基础，结合自己10多年来在工作中遇到的实际代码场景，介绍了数据科学中 Python 的进阶内容。一、数据结构和算法
Python内置了许多非常有用的数据结构，比如列表（list）、集合（set）以及字典（dictionary）。就绝大部分情况而言，我们可以直接使用这些数据结构。但是，通常我们还需要考虑比如搜索、排序、排列以及筛选等这一类常见的问题。因此，本章的目的就是来讨论常见的数据结构和同数据有关的算法。此外，在 collections 模块中也包含了针对各种数据结构的解决方案。
1. 将序列分解为单独的变量
问题
现在有一个包含 N 个元素的元组或序列，怎样将它里面的值分解并赋值给 N 个单独的变量？
解决方案
任何序列（或可迭代的对象）都可以通过一个简单的赋值操作来分解为单独的变量。唯一的要求是变量的总数和结构要与序列相吻合。
示例代码（字符串 string）：
>>> data = "Hello"
>>> a, b, c, d, e = data
>>> a
H
>>> b
e
>>> c
l
>>> d
l
>>> e
o
示例代码（元组 tuple）：
>>> p = (4, 5)
>>> x, y = p
>>> x
4
>>> y
5
示例代码（集合 set）：不建议使用此方法，因为集合（set）是一个无序的不重复元素序列，实际得到的结果和我们想要的结果会存在偏差。
>>> data = set(["ACME", 50, 91.1, (2022, 5, 3)])
>>> name, shares, price, date = data
>>> name
(2022, 5, 3)
>>> shares
50
>>> price
91.1
>>> date
(2022, 12, 21)
示例代码（列表 list）：
>>> data = ["ACME", 50, 91.1, (2022, 5, 3)]
>>> name, shares, price, date = data
>>> name
ACME
>>> shares
50
>>> price
91.1
>>> date
(2022, 12, 21)
示例代码（字典 dictionary）：
>>> data = {"Name": "mahua", "Age": 18, "Gender": "male"}
>>> name, age, gender = data.items()
>>> name
('Name', 'mahua')
>>> age
('Age', 18)
>>> gender
('Gender', 'male')
示例代码（迭代器 iterator）：
>>> data = iter(["ACME", 50, 91.1, (2022, 5, 3)])
>>> name, shares, price, date = data
>>> name
ACME
>>> shares
50
>>> price
91.1
>>> date
(2022, 5, 3)
示例代码（生成器 generator）：
def generator():
for i in ["ACME", 50, 91.1, (2022, 5, 3)]:
yield i
f = generator() # f 是一个迭代器，由生成器返回生成
>>> name, shares, price, date = f
>>> name
ACME
>>> shares
50
>>> price
91.1
>>> date
(2022, 5, 3)
如果元素的数量不匹配，将得到一个错误提示。
示例代码（元组 tuple）：
>>> p = (4, 5)
>>> x, y, z = p
---------------------------------------------------------------------------
ValueError Traceback (most recent call last)
Input In [1], in <cell line: 2>()
1 p = (4, 5)
----> 2 x, y, z = p
ValueError: not enough values to unpack (expected 3, got 2)
讨论
当做分解操作时，有时候可能想丢弃某些特定的值。Python 并没有提供特殊的语法来实现这一点，但是通常可以选择一个用不到的变量名，以此来作为要丢弃的值的名称。（注意：必须确保选择的变量名没有在其他地方用到过。）
示例代码（列表 list）：
>>> data = ["ACME", 50, 91.1, (2022, 5, 3)]
>>> _, shares, price, _ = data
>>> shares
50
>>> price
91.1
>>> data = ["ACME", 50, 91.1, (2022, 5, 3)]
>>> name, shares, *_ = data # 注意，使用 *(星号) 表达式时，需要确定 头部/尾部 变量名
>>> name
ACME
>>> shares
50
>>> data = ["ACME", 50, 91.1, (2022, 5, 3)]
>>> name, *_, date = data # # 注意，使用 *(星号) 表达式时，需要确定 头部/尾部 变量名
>>> name
ACME
>>> date
(2022, 5, 3)
2. 从任意长度的可迭代对象中分解元素
问题
需要从某个可迭代对象中分解出 N 个元素，但是这个可迭代对象的长度可能超过 N，这会导致出现 “分解的值过多（too many values to unpack）” 的异常。
解决方案
Python 的 “*号表达式” 可以用来解决这个问题。由 * 修饰的变量可以位于列表的第一个位置。
例 1：
假设开设了一门课程，并决定在期末的作业成绩中去掉第一个和最后一个分数，只对中间剩下的成绩做平均分统计。如果只有 4 个成绩，也许可以简单地将 4 个都分解出来，但是如果有 24 个呢？*号表达式 使得这一切都变得简单：
from statistics import fmean
def drop_first_last(grades):
first, *middle, last = grades
return middle
>>> grade_list = [100, 98, 99, 87, 64, 82, 71]
>>> average_score = fmean(drop_first_last(grade_list))
>>> average_score
86.0
例 2：
假设有一些用户记录，记录由姓名和电子邮件地址组成，后面跟着任意数量的电话号码。则可以这样分解记录：
>>> record = ("Mahua", "zhinengmahua@163.com", "773-555-1212", "847-555-1212")
>>> name, email, *phone_numbers = record
>>>> name
Mahua
>>> email
zhinengmahua@163.com
>>> phone_numbers
['773-555-1212', '847-555-1212']
注意：不管需要分解出多少个电话号码（0 个或多个），变量 phone_numbers 永远都是列表类型。如此一来，对于任何用到了变量 phone_numbers 的代码都不需要对其进行类型检查。
例 3： * 修饰的变量可以位于列表的第一个位置
假设用一系列的值来代表公司过去 8 个季度的销售额，现在需要对近一个季度的销售额同前 7 个季度的销售额平均值作比较。则可以这样分解：
def avg_comparison(avg, current):
return current - avg
>>> sales_record = [10, 8, 7, 1, 9, 5, 10, 3]
>>> *trailing, current = sales_record # * 修饰的变量位于列表的第一个位置
>>> trailing_avg = sum(trailing) / len(trailing)
>>> result = avg_comparison(trailing_avg, current)
>>> trailing
[10, 8, 7, 1, 9, 5, 10]
>>> current
3
>>> trailing_avg
7.142857142857143
>>> result
-4.142857142857143
讨论
对于分解未知或任意长度的可迭代对象，这种扩展的分解操作可谓是量身定做的工具。
通常，这类可迭代对象中会有一些已知的组件或模式（例如，元素 1 之后的所有内容都是电话号码），利用 * 表达式分解可迭代对象使得开
发者能够轻松利用这些模式，而不必在可迭代对象中做复杂花哨的操作才能得到相关的元素。
(1) * 表达式的语法可用于迭代一个可变长度元组序列。
例如，假设有一个带标记的元组序列：
records = [("foo", 1, 2), ("bar", "hello"), ("foo", 3, 4)]
def do_foo(x, y):
print("foo", x, y)
def do_bar(s):
print("bar", s)
for tag, * args in records:
if tag == "foo":
do_foo(*args)
elif tag == "bar":
do_bar(*args)
# 运行结果：
foo 1 2
bar hello
foo 3 4
(2) * 表达式的语法支持和某些特定的字符串处理操作相结合。
例如，做拆分（splitting）操作：
>>> line = "nobody:*:-2:-2:Unprivileged User:/var/empty:/usr/bin/false"
>>> uname, *fields, homedir, sh = line.split(":")
>>> uname
nobody
>>> fields
['*', '-2', '-2', 'Unprivileged User']
>>> homedir
/var/empty
>>> sh
/usr/bin/false
(3) * 表达式的语法可用于分解出某些值然后丢弃它们。
注意：在分解的时候，不能只是指定一个单独的 *，但是可以使用几个常用来表示待丢弃值的变量名，比如 _ 或者 ign(ignored)。
>>> record = ("ACME", 50, 123.45, (5, 4, 2022))
>>> name, *_, (*_, year) = record
>>> name
ACME
>>> year
2022
(4) * 表达式的语法和各种函数式语言中的列表处理功能有着一定的相似性。
例如，将一个列表分解为头部和尾部：
>>> items = [1, 10, 7, 4, 5, 9]
>>> head, *tail = items
>>>> head
1
>>> tail
[10, 7, 4, 5, 9]
3. 保留最后 N 个元素
问题
在迭代或是其他形式的处理过程中，如何对最后几项记录做一个有限的历史记录统计？
解决方案
使用 collections.deque 实现保留有限的历史记录。
例 1：
对一系列文本做简单的文本匹配操作，当发现有匹配时就输出当前的匹配行以及最后检查过的 N 行文本。
lines = """两只老虎，两只老虎；
跑得快，跑得快；
一只没有眼睛，一只没有尾巴；
真奇怪！真奇怪！
两只老虎，两只老虎；
跑得快，跑得快；
一只没有耳朵，一只没有尾巴；
真奇怪！真奇怪！"""
from collections import deque
def search(lines, pattern, history=5):
previous_lines = deque(maxlen=history)
for line in lines.split("\n"):
if pattern in line:
yield line, previous_lines
previous_lines.append(line)
for line, previous_lines in search(lines, "尾巴", 5):
for previous_line in previous_lines:
print(previous_line, end=" ")
print(line, end=" ")
print("-" * 20)
# 运行结果：
两只老虎，两只老虎； 跑得快，跑得快； 一只没有眼睛，一只没有尾巴； --------------------
跑得快，跑得快； 一只没有眼睛，一只没有尾巴； 真奇怪！真奇怪！ 两只老虎，两只老虎； 跑得快，跑得快； 一只没有耳朵，一只没有尾巴； --------------------
讨论
当编写搜索某项纪录的代码时，通常会用到含有 yield 关键字的生成器函数。这将处理搜索过程的代码和使用搜索结果的代码成功解耦开来。
(1) 有界限队列
deque(maxlen=N) 创建了一个固定长度的队列。当有新纪录加入而队列已满时会自动移除最老的那条记录。
示例代码：
>>> from collections import deque
>>> q = deque(maxlen=3)
>>> q.append(1)
>>> q.append(2)
>>> q.append(3)
>>> q
deque([1, 2, 3], maxlen=3)
>>> q.append(4)
deque([2, 3, 4], maxlen=3)
>>> q.append(5)
deque([3, 4, 5], maxlen=3)
注意：尽管可以手动在列表（list）中实现这一操作（append 、 del），但队列的解决方案要更加优雅、运行速度也快得多。
(2) 无界限队列
deque(maxlen=None) 创建了一个可以增长到任意长度的队列。可以在两端执行添加和弹出操作。
示例代码：
>>> from collections import deque
>>> q = deque(maxlen=None) # q = deque()
>>> q.append(1)
>>> q.append(2)
>>> q.append(3)
>>> q
deque([1, 2, 3])
>>> q.appendleft(4)
>>> q
deque([4, 1, 2, 3])
>>> q.pop()
3
>>> q
deque([4, 1, 2])
>>> q.popleft()
4
>>> q
deque([1, 2])
注意：从队列两端添加或弹出元素的复杂度都是 O(1)。区别于列表，当从列表头部插入或移除元素时，列表的复杂度为 O(N)。
4. 查找最大或最小的 N 个元素
问题
如何从某个集合中找出最大或最小的 N 个元素？
解决方案
heapq 模块中有两个函数：nlargest() 和 nsmallest() 可以完美解决这个问题。
函数 nlargest() 和 nsmallest() 都可以接受一个参数 key，从而允许它们工作在更加复杂的数据结构之上。
例 1：
在一个列表中分别找出最大和最小的 3 个元素。
import heapq
nums = [1, 8, 2, 23, 7, -4, 18, 23, 42, 37, 2]
print(f"最大的 3 个元素列表：{heapq.nlargest(3, nums)}")
print(f"最小的 3 个元素列表：{heapq.nsmallest(3, nums)}")
# 运行结果：
最大的 3 个元素列表：[42, 37, 23]
最小的 3 个元素列表：[-4, 1, 2]
例 2：
在一个包含字典（dictionary）的列表（list）中分别找出价格（price）最高和最低的 3 个子字典（dictionary）。
import heapq
portfolio = [
{'name': 'IBM', 'shares': 100, 'price': 91.1},
{'name': 'AAPL', 'shares': 50, 'price': 543.22},
{'name': 'FB', 'shares': 200, 'price': 21.09},
{'name': 'HPQ', 'shares': 35, 'price': 31.75},
{'name': 'YHOO', 'shares': 45, 'price': 16.35},
{'name': 'ACME', 'shares': 75, 'price': 115.65}
]
cexpensive = heapq.nlargest(3, portfolio, key=lambda s: s["price"])
cheap = heapq.nsmallest(3, portfolio, key=lambda s: s["price"])
print(f"price 最高的 3 个子字典列表：{cexpensive}")
print(f"price 最低的 3 个子字典列表：{cheap}")
# 运行结果：
price 最高的 3 个子字典列表：[{'name': 'AAPL', 'shares': 50, 'price': 543.22}, {'name': 'ACME', 'shares': 75, 'price': 115.65}, {'name': 'IBM', 'shares': 100, 'price': 91.1}]
price 最低的 3 个子字典列表：[{'name': 'YHOO', 'shares': 45, 'price': 16.35}, {'name': 'FB', 'shares': 200, 'price': 21.09}, {'name': 'HPQ', 'shares': 35, 'price': 31.75}]
讨论
(1) 通过 heapq 模块的 heapify() 函数将列表在线性时间内原地转化成堆后，使用 heapq 模块的 heappop() 方法获取最小的元素。
堆最重要的特性就是 heap[0] 总是最小的那个元素。接下来的元素可依次通过 heapq.heappop() 方法获取。
heapq.heappop() 方法会将第一个元素（最小的）弹出，然后以第二小的元素取代被弹出元素（这个操作的复杂度是 O(logN)，N 代表堆的大小）。
示例代码：找到第 3 小的元素。
>>> import heapq
>>> nums = [1, 8, 2, 23, 7, -4, 18, 23, 42, 37, 2, 15, 14, 28, 70, 4, -61]
>>> heap = list(nums)
>>> heapq.heapify(heap) # heapq.heapify(x)：将 list x 转换成堆，原地，线性时间内
>>> heap # 在底层将数据结构转化为列表，且元素以堆的顺序排序
[-61, 1, -4, 4, 2, 2, 18, 8, 42, 37, 7, 15, 14, 28, 70, 23, 23]
>>> heapq.heappop(heap)
-61
>>> heapq.heappop(heap)
-4
>>> heapq.heappop(heap)
1
适用场景：寻找最大或最小的 N 个元素，且同集合中元素的总数相比，N 很小。
(2) 通过 max() 和 min() 函数寻找最大和最小的元素（N=1）。
>>> nums = [1, 8, 2, 23, 7, -4, 18, 23, 42, 37, 2, 15, 14, 28, 70, 4, -61]
>>> max(nums) # 获取最大值
70
>>> min(nums) # 获取最小值
-61
适用场景：寻找最大或最小的 N 个元素，且 N =1。
(3) 通过先对集合进行排序，然后做切片操作获取最大和最小的 N 个元素。
>>> nums = [1, 8, 2, 23, 7, -4, 18, 23, 42, 37, 2, 15, 14, 28, 70, 4, -61]
>>> sorted(nums)[:(len(nums) - 2)] # 获取最小的（集合大小-2）个元素
[-61, -4, 1, 2, 2, 4, 7, 8, 14, 15, 18, 23, 23, 28, 37]
>>> sorted(nums)[-(len(nums) - 2):] # # 获取最大的（集合大小-2）个元素
[1, 2, 2, 4, 7, 8, 14, 15, 18, 23, 23, 28, 37, 42, 70]
适用场景：寻找最大或最小的 N 个元素，且 N 和集合本身的大小差不多大。
5. 实现优先级队列
问题
如何实现一个按优先级排序的队列？并且在对该队列进行 pop 操作时都会返回优先级最高的那个元素。
解决方案
利用 heapq 模块实现一个简单的优先级队列。
例 1：
利用 heapq 模块实现一个简单的优先级队列：
import heapq
class PriorityQueue():
def __init__(self):
self._queue = []
self._index = 0
def push(self, item, priority):
# heapq.heappush(heap, item)：将 item 的值加入 heap 中，保持堆的不变性
heapq.heappush(self._queue, (-priority, self._index, item))
self._index += 1
def pop(self):
# heapq.heappop(heap)：弹出并返回 heap 的最小的元素，保持堆的不变性。如果堆为空，抛出 IndexError 。
# 使用 heap[0] ，可以只访问最小的元素而不弹出它。
return heapq.heappop(self._queue)[-1]
class Item():
def __init__(self, name):
self.name = name
def __repr__(self):
return "Item({!r})".format(self.name)
q = PriorityQueue()
print(q._queue) # []
q.push(Item("foo"), 1)
print(q._queue) # [(-1, 0, Item('foo'))]
q.push(Item("bar"), 5)
print(q._queue) # [(-5, 1, Item('bar')), (-1, 0, Item('foo'))]
q.push(Item("spam"), 4)
print(q._queue) # [(-5, 1, Item('bar')), (-1, 0, Item('foo')), (-4, 2, Item('spam'))]
q.push(Item("grok"), 1)
print(q._queue) # [(-5, 1, Item('bar')), (-1, 0, Item('foo')), (-4, 2, Item('spam')), (-1, 3, Item('grok'))]
print(q.pop()) # Item('bar')
print(q._queue) # [(-4, 2, Item('spam')), (-1, 0, Item('foo')), (-1, 3, Item('grok'))]
print(q.pop()) # Item('spam')
print(q._queue) # [(-1, 0, Item('foo')), (-1, 3, Item('grok'))]
print(q.pop()) # Item('foo')
print(q._queue) # [(-1, 3, Item('grok'))]
print(q.pop()) # Item('grok')
print(q._queue) # []
讨论
