使用itertools.groupby实现排名排序

table {
border-collapse: collapse;
width: 100%;
margin-bottom: 1rem;
}
th, td {
border: 1px solid #ddd;
padding: 8px;
text-align: left;
}
th {
background-color: #f2f2f2;
}
tr:nth-child(even) {
background-color: #f9f9f9;
}
pre {
background-color: #f8f8f8;
padding: 15px;
border-radius: 4px;
overflow-x: auto;
}

17、排名排序。对一组数据中的项进行排名，构建一个具有相同键值的项的字典，使得创建一个作为各项平均值的排名成为可能。例如，序列[0.8, 1.2, 1.2, 2.3, 18]的排名值应为1、2.5、2.5、4、5。序列中位置1和2的两个匹配键值应将中点值2.5作为它们的共同排名。这可以使用itertools.groupby()函数来计算。每个组将有一定数量的成员，由groupby()函数提供。具有匹配键的n个项的组的排名值序列为r0、r0 + 1、r0 + 2、…、r0 + n。r0的值是该组的起始排名。这个序列的平均值是r0 + n/2。此处理需要创建一个临时值序列，以便为具有相同键的组中的每个项及其匹配的排名发出值。请使用itertools.groupby()函数编写这个rank()函数，并说明使用itertools.groupby()实现该功能有哪些优点。

以下是使用

itertools.groupby()

实现

rank()

函数的示例代码：

import itertools

def rank(data):
    sorted_data = sorted(data)
    result = []
    rank_counter = 1
    for key, group in itertools.groupby(sorted_data):
        group_list = list(group)
        group_size = len(group_list)
        mean_rank = rank_counter + (group_size - 1) / 2
        for _ in group_list:
            result.append(mean_rank)
        rank_counter += group_size
    return result

# 测试
data = [0.8, 1.2, 1.2, 2.3, 18]
print(rank(data))

与手动实现比较，

itertools

变体的优点：

代码简洁

：使用

itertools.groupby()

减少了手动构建字典和处理分组逻辑的代码量，使代码更易读和维护。

性能优化

：

itertools.groupby()

是用C实现的，通常比手动编写的Python代码更快，尤其是在处理大数据集时。

代码复用

：

itertools.groupby()

是标准库函数，可在不同场景中复用，提高了代码的可复用性。

18、实现对图像像素进行两次遍历的算法：1. 创建一个计数器统计每种颜色的频率。2. 对于频率低于某个阈值 � 的颜色，定位其相邻像素。在角落处，相邻像素可能少至 3 个；在中间位置，相邻像素最多 8 个。找出这些相邻像素中占多数的颜色并替换该异常颜色。考虑罕见颜色相邻的边缘情况。该算法复杂度对有 1000 万像素的图片是否有帮助？

对于有 1000 万像素的图片，该算法复杂度可能并不高效。算法需遍历所有像素统计颜色频率，再遍历低频颜色像素并处理相邻像素，计算量和时间复杂度较高。

不过若罕见颜色少：

算法能有效减少异常颜色

优化图像颜色分布

若罕见颜色多且相邻：

边缘情况处理复杂

会增加计算量和复杂度，效率降低

19、定义并测试一个容错的字符串到十进制数的转换函数。该函数应去除美元符号和逗号，并将剩余的字符串转换为 decimal.Decimal 类型。

from decimal import Decimal

def tolerant_str_to_decimal(text: str | None) -> Decimal | None:
    if text is None:
        return None
    return Decimal(text.replace("$", "").replace(",", ""))

# 测试示例
test_str = "$1,234.56"
result = tolerant_str_to_decimal(test_str)
print(result)

上述代码定义了

tolerant_str_to_decimal

函数，它接收一个字符串参数，去除其中的美元符号和逗号后，将剩余的字符串转换为

Decimal

类型。如果输入为

None

，则返回

None

。最后给出了一个简单的测试示例。

20、第一个需求是编写一个函数，用于计算字符串数据的均值、方差和标准差，公式分别为：均值(�) = ∑�∈�� / len(�)；方差(�) = ∑�∈�(� – 均值(�))² / (len(�) – 1)；标准差(�) = √方差(�)。使用 @cache 装饰器定义一个函数（该函数将字符串数据转换为浮点数，可命名为 str_to_float()），以缓存该函数的中间数值结果。创建一个非常大的随机数字符串集合，比较使用缓存和不使用缓存这两种方案计算均值、方差和标准差的速度。

需要编写计算字符串数据均值、方差和标准差的函数，使用 `@cache` 装饰器缓存字符串转浮点数函数的中间结果，创建大的随机数字符串集合并比较使用缓存和不使用缓存这两种方案计算均值、方差和标准差的速度。

21、设计方案一：将经过清理的中间值具体化，创建一个仅包含纯数值的临时序列对象，然后对这些纯数值列表计算各种统计指标。有不同的缓存方案，有一种方法是使用sys.getallocatedblocks()来了解Python使用了多少内存。请通过该方法确定哪种缓存方案使用的内存最少，并展示结果，以说明哪种设计方案在性能和内存使用方面是最佳的。

可以通过创建仅含纯数值的临时序列对象来计算统计指标，利用

sys.getallocatedblocks()

方法评估不同缓存方案的内存使用情况，通过比较不同设计方案在性能和内存使用上的结果，确定最佳方案。

22、在计算分数之和时，原本使用

/

真除法运算符，该运算符在

operator

模块中对应

operator.truediv()

函数，此函数会创建

float

对象。现在要求用

fractions.Fraction()

函数（会创建

Fraction

对象）替换

operator.truediv()

函数，这样能创建精确的有理数值，避免浮点数近似的问题。请更改此运算符，并验证求和结果仍然近似于π。

大致来说，需要把使用

operator.truediv()

函数计算分数的代码部分，替换为

fractions.Fraction()

函数进行计算，然后验证求和结果仍近似于 π。

23、有一个文本文件Anscombe.txt，其中包含四个(x, y)对的系列。每个系列都有相同的已知均值和标准差，但每个系列差异很大，需要四个不同的模型来计算给定x值对应的预期y值。数据以表格形式呈现，第一行是标题，第二行是系列名称，第三行是每个系列的两列名称，其余行是每个系列的x和y值。需要将这些数据分解为四个单独的序列，每个序列包含键为“x”和“y”的二元字典。解析的基础是csv模块，它会将每行转换为字符串序列，但每个序列中有来自四个不同系列的八个样本。可以使用toolz.itertoolz或itertools来完成剩余的分解工作。编写一个解析器来分解Anscombe数据集，并将值从字符串转换为浮点值，以便为每个系列计算描述性统计信息。

以下是一个使用Python的

csv

模块和

toolz.itertoolz

库来分解Anscombe数据集的示例代码：

import csv
from toolz.itertoolz import partition
from pathlib import Path

def row_iter(source):
    return csv.reader(source, delimiter="	")

def parse_anscombe_data(file_path):
    with Path(file_path).open() as source:
        # 跳过前三行非数据行
        next(source)
        next(source)
        next(source)
        data = []
        for row in row_iter(source):
            # 将每行数据拆分为四个系列的(x, y)对
            series_data = partition(2, map(float, row))
            for x, y in series_data:
                data.append({'x': x, 'y': y})
        # 分解为四个单独的序列
        series1 = data[::4]
        series2 = data[1::4]
        series3 = data[2::4]
        series4 = data[3::4]
        return series1, series2, series3, series4

# 使用示例
file_path = "Anscombe.txt"
series1, series2, series3, series4 = parse_anscombe_data(file_path)
print(series1)
print(series2)
print(series3)
print(series4)

这段代码首先定义了一个

row_iter

函数，用于将文件按行读取并使用

csv

解析。

parse_anscombe_data

函数用于处理文件，跳过前三行非数据行，将每行数据拆分为四个系列的

(x, y)

对，并将其转换为浮点值。最后，将数据分解为四个单独的序列并返回。

24、有一个包含长途旅行一系列航点信息的文件，文件中每个航点信息可通过一个函数转换为字符串序列。现在有一个 distance() 函数，用于计算两个航点之间的距离，该函数原本是为处理复杂的 NamedTuple 对象而设计的。要求重新设计并重新实现这个 distance 函数，使其能够处理用包含“latitude”和“longitude”键的字典表示的点。同时，重新设计源数据解析，使用 toolz 包，通过 toolz.functoolz.pipe 将源字符串转换为更有用的结果字典，并确保将纬度和经度值转换为带正确符号的浮点值。最后，比较和重新设计前后的两种实现，从清晰简洁程度方面进行对比，并使用 timeit 模块比较性能，看看是否有特定的性能优势。

本题要求对航点数据处理进行一系列操作。首先需重新设计并实现

distance

函数，使其能处理用含

"latitude"

和

"longitude"

键的字典表示的点；接着重新设计源数据解析，利用

toolz

包和

toolz.functoolz.pipe

将源字符串转换为有用的字典，并把纬度和经度值转换为浮点值。最后比较新旧两种实现方式，从清晰简洁程度和性能两方面进行评估，性能评估可借助

timeit

模块。

25、航点计算中，将航点视为具有纬度和经度的独立位置样本，通过最大和最小纬度以及最大和最小经度可计算出一个边界。实际上，越靠近北极，该区域是一种越靠近极点越窄的梯形。现在需要构建一个类似于航点计算练习中描述的解析管道，找出每个轴上的极值，以定义一个围绕整个航程的框。有几种界定航程的方法：一是给定航程的边界，为边界梯形的四个角定义四个点，用这四个点确定旅程的中点；二是给定两个纬度和经度序列，计算平均纬度和平均经度；三是给定两个纬度和经度序列，计算中位纬度和中位经度。在知道了边界和中心的替代方案后，使用等矩形距离计算来确定旅程中最接近中心的点。请总结上述航点计算及航程界定的主要内容。

在航点分析中，可把航点看作独立位置样本，通过最大最小经纬度确定边界，此边界实际是靠近北极变窄的梯形。

需构建解析管道，找出各轴极值确定航程范围。

界定航程的方法有：

用边界定义梯形四角点来确定中点

根据经纬度序列计算平均经纬度

根据经纬度序列计算中位经纬度

确定边界和中心后，用等矩形距离计算找出旅程中最接近中心的点。

26、编写一个装饰器，用于处理参数为浮点数并返回浮点数的可调用函数，使其能优雅地处理 None 值。若该装饰器名为 @none_tolerant，以下是一个测试用例：使用 @none_tolerant 装饰函数 x2，该函数接受一个浮点数参数并返回其两倍。测试函数 test_x2 中，验证 x2(42.) 等于 84.0，x2(None) 等于 None，以及对列表 [1, 2, None, 3] 应用 x2 后的结果为 [2, 4, None, 6]。

以下是实现

@none_tolerant

装饰器的代码：

from collections.abc import Callable
from functools import wraps

def none_tolerant(function: Callable[[float], float]) -> Callable[[float | None], float | None]:
    @wraps(function)
    def null_wrapper(value: float | None) -> float | None:
        return None if value is None else function(value)
    return null_wrapper

@none_tolerant
def x2(x: float) -> float:
    return 2 * x

def test_x2() -> None:
    assert x2(42.) == 84.0
    assert x2(None) is None
    assert list(map(x2, [1, 2, None, 3])) == [2, 4, None, 6]
    print("All tests passed!")

test_x2()

在上述代码中，定义了

none_tolerant

装饰器，它接受一个函数作为参数，并返回一个新的函数

null_wrapper

。

null_wrapper

函数会检查传入的值是否为

None

，如果是则返回

None

，否则调用原函数。然后使用该装饰器装饰

x2

函数，并编写了

test_x2

测试函数进行验证。

27、请完成以下任务：首先，定义一些函数来复制目录。需要单独定义函数完成以下状态变化：创建一个新的空目录；将文件从源目录的某个位置复制到目标目录。可以使用 pathlib.Path.glob() 方法提供目录内容的视图，进而编写一个整体函数来调用另外两个函数创建子目录并复制文件。其次，定义一个装饰器，在进行试运行时阻止操作，并将该装饰器应用于目录创建函数和文件复制函数，注意这两个函数的签名不同，一个函数使用单个路径，另一个函数使用两个路径。第三，创建一个合适的单元测试，以确认试运行模式只是走过场，不会改变底层文件系统。可以使用 pytest.tmp_path 夹具提供一个临时工作目录，避免在调试时不断删除和重新创建输出目录。

目录复制功能开发步骤说明

以下是实现目录复制功能的开发步骤：

定义函数


分别编写以下两个函数：

– 创建新空目录的函数。

– 从源目录复制文件到目标目录的函数。

可借助

pathlib.Path.glob()

方法辅助实现整体复制功能。

定义装饰器


创建一个装饰器，根据试运行状态决定是否执行函数操作。

将该装饰器应用于目录创建和文件复制函数。

编写单元测试


使用

pytest.tmp_path

夹具创建临时工作目录。

编写单元测试，验证试运行模式不会改变底层文件系统。

28、编写

non_excluded_names()

和

non_excluded_ext()

函数以完成

path_filter()

函数的实现。其中，’favicon.ico’ 和 ‘robots.txt’ 这样的名称需要被排除，’.js’ 和 ‘.css’ 这样的扩展名也需要被排除。这些函数以及整体的

path_filter()

函数都需要处理分解后的路径名，并且要编写相应的单元测试用例。思考尝试为这三个操作（路径名排除、扩展名排除、整体路径过滤）编写一个单一的复杂函数是否明智，将三个单独的规则分解并通过一个整体的路径过滤函数进行组合是否更有意义？

为完成

path_filter()

函数，可使用以下两个函数进行路径过滤判断：

non_excluded_names()

函数用于检查路径中的文件名是否为

'favicon.ico'

或

'robots.txt'

等，若不是则返回

True

。

non_excluded_ext()

函数用于检查文件的扩展名是否为

.js

或

.css

等，若不是则返回

True

。

关于是否编写单一复杂函数：

编写一个功能集中但复杂的函数虽然可以将所有逻辑集中管理，但会使代码难以维护和扩展。相比之下，将规则进行分解并通过组合的方式实现更具灵活性，便于后续添加新的过滤规则以及进行单元测试。

因此，

将规则分解并组合到整体路径过滤函数中是更合理的方式

。

29、在分析大量数据时，原本设计是为每个文件分配一个单独的工作进程。现考虑以下替代方案：一是创建两个工作进程池，一个池读取文件并以1024行为一组返回行，第二个池包含analysis()函数主要部分，解析每行创建Access对象、AccessDetails对象，应用过滤器并汇总结果；二是将10 Mb的日志文件分解为更小文件，每个新文件限制为4096个单独的日志条目，再将分析应用程序应用于这些小文件集合；三是将analysis()函数分解为使用三个单独的工作进程池，一个池解析文件返回Access对象块，另一个池将Access对象转换为AccessDetails对象，第三个池应用过滤器并汇总AccessDetails对象。请总结使用不同处理管道分析大量数据的结果。

数据处理优化设计

在使用不同处理管道分析大量数据时，可以尝试以下多种替代设计：

创建两个工作进程池

将大日志文件分解为小文件

将分析函数分解为三个工作进程池

设计关注点

在设计过程中，应重点关注以下几个方面：

将I/O处理分配到尽可能多的CPU或核心上

最小化进程间传递的数据量

通过基准测试实验确定资源的最优分配和设计

初步测试结果

不同方法的处理时长存在差异，具体结果如下：

30、在将结果数据序列化为 CSV 和 JSON 的基础上，还要将其序列化为 XML，有哪些实现方法，并且还需要做什么来确保结果符合要求？

可以通过两种方式实现将结果数据除 CSV 和 JSON 外还序列化为 XML。一是下载安装能序列化

Series

和

Pair

对象的库；二是编写可处理

list[dict[str, Any]]

对象的函数。同时要添加测试用例，确保响应格式和内容符合预期。

31、在已有将结果数据序列化为 CSV 和 JSON 的示例中，扩展响应，将结果数据也序列化为 HTML。已知 HTML 序列化比 XML 序列化更复杂，因为 HTML 数据呈现存在较多开销。通常做法是包含一个完整的 HTML 表格结构来映射 CSV 的行和列，而不是表示 Pair 对象。同时，添加 HTML 序列化格式时需要添加测试用例，以确认响应具有预期的格式和内容。请说明具体的操作步骤。

在已有将数据序列化为 CSV 和 JSON 的示例基础上，进一步把结果数据序列化为 HTML。由于 HTML 数据呈现开销大，其序列化比 XML 更复杂。通常要包含完整 HTML 表格结构来对应 CSV 的行和列，而非仅表示 Pair 对象。并且要添加测试用例，确保响应的格式和内容符合预期。

32、为每个样本创建一个命名元组（NamedTuple），并编写一些函数以有用的形式获取这些数据。定义一个相关函数，将此相关函数应用于发动机和转速计的值，以查看这些值是否相关。如果相关，则表明发动机的仪器可以重新校准；如果不相关，则表明发动机存在其他问题。如有必要，重写类型提示或命名元组定义。

先创建命名元组和获取数据的函数，再定义相关函数，将其应用于发动机和转速计的值判断相关性，若有必要重写类型提示或命名元组定义。

33、将 make_color_map() 函数重新定义为一个高阶函数，使其能够接受一个距离函数作为参数。所有距离函数的类型提示应为 Callable[[RGB, Color], float]。一旦 make_color_map() 函数被修改，就可以用不同的距离函数创建不同的颜色映射。选择不同的距离函数会产生多大的差异？如何描述从大量可能的 RGB 值到有限子集颜色域的映射？展示有多少不同的 RGB 值映射到一个子集颜色的直方图是否合理且有信息量？

不同距离函数的计算时间对比

选择不同的距离函数在计算时间上有明显差异：

欧几里得距离

：计算 100 万次需

1.761 秒

曼哈顿距离

：计算 100 万次需

0.857 秒

当计算次数扩展到

10 亿次

时：

曼哈顿距离约需

半小时

欧几里得距离约需

46 分钟