Метка: python

Стратегия на Python

​​«Расскажите про любой шаблон проектирования на ваш выбор.»

Случалось слышать такое на собеседованиях? Большинство людей в этот момент начинают рассказывать про синглтон (одиночку). Потому что он… простой? Да, вообще-то не очень. Попробуйте сходу вспомнить, как там реализовать его через метакласс. Да и часто ли приходится? Скорее всего вы пользуетесь уже готовым кодом для синглтона. Его даже называют «анти-паттерном», потому что он часто маскирует плохой дизайн кода, вызывает проблемы при тестировании и нарушает принцип единственной отвественности класса (и порождает себя, и делает какую-то работу). А еще, он может вызывать проблемы с многопоточностью или «многопроцессностью» в случае с Python. Поэтому хвастать знанием синглотона – не лучшая стратегия на собеседовании…

Стратегия выбора работника

Ага! Стратегия! Это именно тот шаблон, который действительно подойдет для рассказа, потому что он простой и реально часто применяется на практике, даже если вы порой сами это не осознаете.

Стратегия – поведенческий шаблон, призванный для обеспечения взаимозаменяемости разных алгоритмов или вариаций алгоритма с одинаковыми интерфейсами. Стратегии – и есть эти варианты. В зависимости от условий (контекст) код выбирает подходящий алгоритм. 

Реализация этого шаблона может быть не только объектная, но и функциональная. С последней и начнем:

# стратегия печатать на экран
def console_writer(info):
    print(info)

# стратегия выводить в файл
def file_writer(info):
    with open('log.txt', 'a') as file:
        file.write(info + '\n')

def client(writer):
    writer('Hello world!')
    writer('Good bye!')

# пользователь выбирает стратегию
if input('Write to file? [Y/N]') == 'Y':
    client(writer=file_writer)
else:
    client(writer=console_writer)

Стратегия выбирается пользователем, а функция client даже не знает, какой вариант алгоритма ей дадут. Она знает лишь то, что writer(info) – это некая функция, принимающая строку (это и есть общий интерфейс для всех стратегий). Таким образом, мы делегируем работу стратегиям, скрывая детали реализации каждой из них.

В объектном варианте:

class Adder:
    def do_work(self, x, y):
        return x + y

class Multiplicator:
    def do_work(self, x, y):
        return x * y

class Calculator:
    def set_strategy(self, strategy):
        self.strategy = strategy

    def calculate(self, x, y):
        print('Result is', self.strategy.do_work(x, y))

calc = Calculator()
calc.set_strategy(Adder())
calc.calculate(10, 20)

calc.set_strategy(Multiplicator())
calc.calculate(10, 20)

Мы обеспечили горячую заменяя алгоритмов для класса Calculator. Для простоты, здесь я не применял наследование (спасибо динамической природе Python), но в серьезных проектах, вам следовало бы написать что-то подобное:

from abc import ABC, abstractmethod

class BaseStrategy(ABC):
    @abstractmethod
    def do_work(self, x, y):
        pass

class Adder(BaseStrategy):
    def do_work(self, x, y):
        return x + y

class Multiplicator(BaseStrategy):
    def do_work(self, x, y):
        return x * y

class Calculator:
    def set_strategy(self, strategy: BaseStrategy):
        self.strategy = strategy

    def calculate(self, x, y):
        print('Result is', self.strategy.do_work(x, y))

Здесь мы создаем общий интерфейс стратегий BaseStrategy – как абстрактный класс ABC. Далее в каждой стратегии реализуем этот интерфейс.

Надеюсь, было полезно. Если хотите еще больше подробностей, то читайте 1, 2, 3.

Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈 

Ключ сортировки key

Методы sort, sorted могут принимать именованный аргумент key. Он должен быть функцией (или чем-то другим вызываемым – callable) с одним аргументом. Смысл key в том, что он вызывается ровно один раз для каждого из элементов списка (итератора и т.п.), которой мы сортируем, и указывает порядок сортировки: элементы выстраиваются ровно в том порядке, в каком бы выстроился сортированный список результатом вызова key на всех элементах:

  1. Применить key ко всем элементам
  2. Отсортировать результаты key по порядку, используя обычное сравнение «больше-меньше» для (чисел, строк и т.п.)
  3. Выстроить исходные данные согласно этому порядку.

key нужен, чтобы либо сортировать данные по нестандартному признаку или нескольким признакам сразу.

С числами все понятно: 10 > 6, 5 < 7. Строки сортируются лексикографически (как статьи в словаре: А < АА < ААА < ААБ < ААВ < АБ < Б < … < ЯЯЯ). А вот сортировка по длине строки потребует использовать key , потому что признак уже нестандартный:

>>> sorted(['Wolf', 'Sparrow', 'Cat'], key=len)
['Cat', 'Wolf', 'Sparrow']

>>> len('Cat')
3
>>> len('Wolf')
4
>>> len('Sparrow')
7

Или другой пример. У нас есть список координат точек [(x, y), ...]. Хотим расположить их по расстоянию от начала координат (0, 0):

pts = [(10, 20), (-100, 150), (0, 0), (40, -30)]

print(sorted(pts, key=lambda p: p[0] ** 2 + p[1] ** 2))

# [(-100, 150), (40, -30), (10, 20), (0, 0)]

Но, что если у нас, скажем, список кортежей? По умолчанию (стандартно) кортежи сравниваются сначала по первому элементу, а потом, если первые равны – по второму, и так далее. Если нужно игнорировать первый элемент и сразу сравнивать по второму – это уже и есть нестандартный признак.

drinks = [
    # напиток, цена
    ('Juice', 100),
    ('Beer', 200),
    ('Soda', 50),
    ('Cocktail', 400),
    ('Water', 20)
]

print(sorted(drinks, key=lambda drink: drink[1]))
# [('Water', 20), ('Soda', 50), ('Juice', 100), ('Beer', 200), ('Cocktail', 400)]


# без key:
print(sorted(drinks))  # отсортирует по названию напитка
# [('Beer', 200), ('Cocktail', 400), ('Juice', 100), ('Soda', 50), ('Water', 20)]

Бонус: если хотите поменять порядок сортировки на обратный, можно либо в лямбде поставить минус перед возвращаемым значениям, но лучше в sorted передать reverse=True.

print(sorted(drinks, key=lambda drink: -drink[1]))
# или лучше
print(sorted(drinks, key=lambda drink: drink[1], reverse=True))
# [('Cocktail', 400), ('Beer', 200), ('Juice', 100), ('Soda', 50), ('Water', 20)]

Сортировка по нескольким признакам сразу

Задача на примере тех же напитков. Отсортировать сначала по числу букв в названии, а если название одинаковой длины, то отсортировать по цене от дорогих к дешевым:

print(sorted(drinks, key=lambda dr: (len(dr[0]), -dr[1])))

# [('Beer', 200), ('Soda', 50), ('Juice', 100), ('Water', 20), ('Cocktail', 400)]

Здесь использовано свойство сравнения кортежей. Давайте прогоним нашу лямбду по элементам и посмотрим на те ключи, которые действительно будет сравнивать sorted:

>>> list(map(lambda dr: (len(dr[0]), -dr[1]), drinks))
[(5, -100), (4, -200), (4, -50), (8, -400), (5, -20)]

Видите, первым теперь идет длина строки, а потом цена с минусом. Поэтому первыми после сортировки пойдут элементы с четверкой в ключе, а среди двух (4, -200), (4, -50) порядок сохранится, потому что -200 < -50.

Модуль operator

Вместо лямбды можно взять одну из библиотечных функций из модуля operator. Есть несколько вариантов для разных ситуаций.

Функция itemgetter(i) берет i-тый элемент кортежа или списка (или ищет по ключу i в dict):

from operator import itemgetter

print(sorted(drinks, key=itemgetter(1)))
# [('Water', 20), ('Soda', 50), ('Juice', 100), ('Beer', 200), ('Cocktail', 400)]

Для словарей:

# преобразуем список кортежей в список словарей
drinks_dict = [{'n': name, 'pr': price} for name, price in drinks]  
print(drinks)

# вывод: [{'n': 'Juice', 'pr': 100}, {'n': 'Beer', 'pr': 200}, {'n': 'Soda', 'pr': 50}, {'n': 'Cocktail', 'pr': 400}, {'n': 'Water', 'pr': 20}]

print(sorted(drinks, key=itemgetter('pr')))

# вывод: [{'n': 'Water', 'pr': 20}, {'n': 'Soda', 'pr': 50}, {'n': 'Juice', 'pr': 100}, {'n': 'Beer', 'pr': 200}, {'n': 'Cocktail', 'pr': 400}]

Теперь представим, что у нас есть класс Drink, и нужно сортировать по атрибуту price. Это можно сделать лямбдой или функцией attrgetter, которая получает атрибут объекта по имени этого атрибута:

class Drink:
    def __init__(self, name, price):
        self.name = name
        self.price = price
    def __repr__(self):
        return f'Drink("{self.name}", {self.price})'

drinks_cls = [
    # напиток, цена
    Drink('Juice', 100),
    Drink('Beer', 200),
    Drink('Soda', 50),
    Drink('Cocktail', 400),
    Drink('Water', 20)
]

print(sorted(drinks_cls, key=lambda drink: drink.price))

# или
from operator import attrgetter
print(sorted(drinks_cls, key=attrgetter('price')))

Еще есть methodcaller, она вызывает метод по имени и использует результат его, как ключ для сортировки.

from operator import methodcaller

sorted(items, key=methodcaller('get_reserve', category='home'))

# тоже самое что:

sorted(items, key=lambda item: item.get_reserve(category='home'))

Исследование производительности

Вариант с лямбдой немного медленнее (потому что операторы написаны на Си, а лямбду – мы пишем на Python). Проведем тесты производительности:

from random import shuffle
from timeit import timeit
from operator import itemgetter

data = [{'ident': x, 'value': 'foo'} for x in range(1000)]
shuffle(data)

def sort_itemgetter(data):
    data.sort(key=itemgetter('ident'))

def sort_lambda(data):
    data.sort(key=lambda it: it['ident'])

print('sort_itemgetter:', timeit('sort_itemgetter(list(data))', globals=globals(), number=10000))
print('sort_lambda:', timeit('sort_lambda(list(data))', globals=globals(), number=10000))

# sort_itemgetter: 1.6157471220000001
# sort_lambda: 1.8793544059999998

Потому что:

ig = itemgetter('ident')
la = lambda it: it['ident']
di = {'ident': 10}

print('itemgetter:', timeit('ig(di)', globals=globals(), number=1000000))
print('lambda:', timeit('la(di)', globals=globals(), number=1000000))

# itemgetter: 0.083
# lambda: 0.11

itemgetter быстрее, чем lambda, ибо он написан на Си.

Смотрите, как вам удобнее. Лично мне нравится все-таки вариант с лямбдами, потому что в нем меньше возможности ошибиться, так как нет строк, зато работает авто-дополнение от среды разработки.

min и max

Методы min и max также поддерживают key. Они вернут соответственно элемент, у которого key вернет наименьшее или наибольшее значение. На примере длины строк:

names = ['Wolf', 'Sparrow', 'Cat']
min(names, key=len)  # 'Cat'
max(names, key=len)  # 'Sparrow'

Самая ближняя от начала координат точка и самая дальняя:

pts = [(10, 20), (-100, 150), (0, 0), (40, -30)]
min(pts, key=lambda p: p[0] ** 2 + p[1] ** 2)  # (0, 0)
max(pts, key=lambda p: p[0] ** 2 + p[1] ** 2)  # (-100, 150)

Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈 

Коронавирус: предсказание на Python

Страшный вирус из Китая ежедневно наращивает свои позиции. Давайте побудем немного британскими учеными и попытаемся спрогнозировать Судный День, когда вся планета будет заражена.

КДПВ: вирус + Китай + график

Нам понадобятся библиотеки:

  • pandas – для загрузки данных
  • matplotlib – для построения графика
  • scipy – для построения предсказания
  • numpy – для работы с массивами

Установите их, если они еще не установлены, набрав в терминале:

pip install pandas matplotlib scipy numpy

Подключим модули:

import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
from scipy.optimize import curve_fit
from datetime import timedelta

Я собрал данные по заболевшим в последние дни в CSV файл и закачал его на свой сайт по адресу: https://tirinox.ru/wp-content/uploads/2020/01/corona.csv

Pandas умеет читать данные прямо из интернета по URL:

df = pd.read_csv('https://tirinox.ru/wp-content/uploads/2020/01/corona.csv', parse_dates=['day'])

print(df.head())

""" Вывод:
         day  infected  dead
0 2020-01-20       278     4
1 2020-01-21       326     6
2 2020-01-22       547     8
3 2020-01-23       639    14
4 2020-01-24       916    25
"""

Мы будем экстраполировать (то есть предсказывать) количество заболевших на будущие дни с учетом текущей динамики. Но какой функцией нам воспользоваться, чтобы она максимально точно вписалась в данные о заболевших? Построим ка график:

df.plot(kind='bar', x='day', y=['infected', 'dead'])
plt.show()
График роста вируса по дням
infected = зараженные люди, dead = погибшие

Очевидно, что прирост с каждым днем все больше и больше. Явно нелинейная зависимость! Это и понятно. Представим, что 1 человек в среднем заражает двоих. Те двое, каждый заразив еще 2, добавят 4 новых человека. А те 4 заразят 8. Затем 16, 32, 64. К этому нужно не забыть добавить уже больных.

Такой закон называется экспоненциальным. Опишем экспоненциальную функцию в параметрическом виде. Параметры a, b, c еще предстоит подобрать, чтобы максимально точно удовлетворить существующим данным:

def fit_func_exp(x, a, b, c):
    return a * np.exp(c * x) + b

Теперь вычислим оптимальные параметры, обеспечивающие наименьшую ошибку:

# зависимая переменная - ее будем предсказывать
infected = df['infected']

# дни - преобразуем их в целые числа от 0 до максимального
days = range(len(df['day']))

# у нас 3 параметра в функции: a, b, c – начальное приближение - единицы
p0 = np.ones(3)

# впишем кривую в наши данные
(a, b, c), _ = curve_fit(fit_func_exp, days, infected, p0=p0)

Теперь, зная параметры, рассчитаем функцию, скажем, до 60 дней с момента начала эпидемии:

# предскажем динамику вируса на 60 дней (начало соответствует 20 января)
MAX_DAY = 60

x_days = np.linspace(0, MAX_DAY - 1, MAX_DAY)
y_infect = fit_func_exp(x_days, a, b, c)

Построим график и убедимся, что он хорошо описывает экспериментальные данные:

plt.xlabel('Дни')
plt.ylabel('Больные')

# график в log шкале
plt.yscale('log')

# это данные текущей статистики - нарисуем их синими точками
plt.scatter(days, infected, marker='D', label='Реальные')

# это красная линия – предсказание (первые 22 дня)
plt.plot(x_days[:22], y_infect[:22], 'r', label='Предсказание')
plt.legend()

plt.show()

Вот, что у нас получится:

График предсказания.

Ого! Рост поражает: сотни, тысячи, десятки тысяч! Узнаем, на какой день число зараженных людей достигнет населения всей Земли:

# население Земли
EARTH_POPULATION = 7_530_000_000

# найдем номер дня, когда количество зараженных достигнет популяции Земли
doom_index = np.argmax(y_infect >= EARTH_POPULATION)
doom_day = x_days[doom_index]

# вычислим дату
day0 = df['day'][0]
doom_date = day0 + timedelta(days=int(doom_day))

# дата конца!
print(f'Doom date: {doom_date:%d.%m.%Y}')

Doom date: 13.03.2020

13 марта…

Полный код загрузил в gist.

UPD: добавил табличку:

+------------+------------------+----------------+
|    Дата    | Число заболевших | Число погибших |
+------------+------------------+----------------+
| 20.01.2020 |        54        |       7        |
| 21.01.2020 |       239        |       9        |
| 22.01.2020 |       492        |       11       |
| 23.01.2020 |       839        |       15       |
| 24.01.2020 |       1314       |       21       |
| 25.01.2020 |       1963       |       30       |
| 26.01.2020 |       2853       |       45       |
| 27.01.2020 |       4072       |       68       |
| 28.01.2020 |       5740       |      104       |
| 29.01.2020 |       8024       |      161       |
| 30.01.2020 |      11150       |      252       |
| 31.01.2020 |      15431       |      394       |
| 01.02.2020 |      21293       |      618       |
| 02.02.2020 |      29317       |      971       |
| 03.02.2020 |      40304       |      1528      |
| 04.02.2020 |      55347       |      2405      |
| 05.02.2020 |      75942       |      3787      |
| 06.02.2020 |      104139      |      5965      |
| 07.02.2020 |      142745      |      9398      |
| 08.02.2020 |      195601      |     14807      |
| 09.02.2020 |      267968      |     23331      |
| 10.02.2020 |      367047      |     36763      |
| 11.02.2020 |      502698      |     57930      |
| 12.02.2020 |      688423      |     91288      |
| 13.02.2020 |      942703      |     143854     |
| 14.02.2020 |     1290844      |     226690     |
| 15.02.2020 |     1767494      |     357228     |
| 16.02.2020 |     2420088      |     562938     |
| 17.02.2020 |     3313572      |     887108     |
| 18.02.2020 |     4536865      |    1397953     |
| 19.02.2020 |     6211706      |    2202971     |
| 20.02.2020 |     8504777      |    3471566     |
| 21.02.2020 |     11644280     |    5470689     |
| 22.02.2020 |     15942657     |    8621023     |
| 23.02.2020 |     21827679     |    13585498    |
| 24.02.2020 |     29885019     |    21408803    |
| 25.02.2020 |     40916535     |    33737214    |
| 26.02.2020 |     56020077     |    53165030    |
| 27.02.2020 |     76698735     |    83780495    |
| 28.02.2020 |    105010434     |   132026095    |
| 29.02.2020 |    143772730     |   208054274    |
| 01.03.2020 |    196843216     |   327863828    |
| 02.03.2020 |    269503423     |   516666580    |
| 03.03.2020 |    368984436     |   814192761    |
| 04.03.2020 |    505186524     |   1283051543   |
| 05.03.2020 |    691664408     |   2021906043   |
| 06.03.2020 |    946976216     |   3186235247   |
| 07.03.2020 |    1296530371    |   5021051835   |
| 08.03.2020 |    1775114218    |   7912460814   |
| 09.03.2020 |    2430356032    |  12468908548   |
| 10.03.2020 |    3327464945    |  19649219634   |
| 11.03.2020 |    4555720506    |  30964364746   |
| 12.03.2020 |    6237357709    |  48795417935   |
| 13.03.2020 |    8539731720    |  76894611953   |
| 14.03.2020 |   11691972927    |  121174929895  |
| 15.03.2020 |   16007789810    |  190954388899  |
| 16.03.2020 |   21916688952    |  300916853606  |
| 17.03.2020 |   30006719188    |  474201998218  |
| 18.03.2020 |   41082993743    |  747274645537  |
| 19.03.2020 |   56247814448    | 1177598150075  |
+------------+------------------+----------------+

Предостережение

Данные расчеты чисто теоретические. Как будет в реальности, мне неизвестно. Но я уверен, что власти Китая и других стран справятся с эпидемией быстрее, чем она поразит весь мир.

Узнавать оперативно о статистике вируса можно через телеграм бот @NovelCoronaVirusBot.

Желаю вам здоровья и не подхватить даже обычной простуды! Меньше бывайте в людных местах, часто мойте руки с мылом, носите маски, при любых симптомах обращайтесь к врачу!

Если заболели, не пытайтесь скрыть болезнь, сбив температуру!

Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈 

Отправить Bitcoin средствами Python

КДПВ

Чтобы создать биткоин транзакцию в наше время не нужно прилагать много усилий. Есть специальные доверенные онлайн сервисы, которые отправят вашу транзакцию в сеть бесплатно (без учета комиссии сети) и безопасно. Вам даже не нужно устанавливать ноду биткоина локально и выкачивать весь блокчейн. Еще лучше то, что под Python есть супер-удобные библиотеки, чтобы пользоваться этими сервисами.

Давайте поставим библиотеку bit:

pip install bit

Ключи

Ключ – центральное понятие в мире Биткоина. Приватный ключ – это ваш кошелек. Вы храните его в секрете от всех. Публичный ключ получается из приватного. А адрес кошелька, получается из публичного ключа. Это преобразование одностороннее: нужно затратить колоссальный объем вычислительной мощности и миллиарды лет времени, чтобы к адресу подобрать приватный ключ и получить контроль над средствами. Я уже рассказывал о генерации ключей биткоин вручную. Но библиотека bit прекрасно делает это за вас.

У биткоина две сети – главная и тестовая. В каждой свои ключи и свои виды адресов. Генерация нового ключа для основной сети, где адреса обычно начинаются с цифры:

from bit import Key

key = Key()
print(k.address)  # 1C8REeQUnXE3HtLCYXVG21AryDxRXnnyar

Класс Key – псевдоним для PrivateKey:

from bit import Key, PrivateKey
print(PrivateKey is Key)  # true

Для демонстрационных целей, я буду использовать тестовую сеть. В ней монеты ничего не стоят и их легко получить. Адреса тестовой сети обычно начинаются с буквы m или n! Ключ тестовой сети описан классом PrivateKeyTestnet:

from bit import PrivateKeyTestnet

k = PrivateKeyTestnet()
print(k.address)  # mrzdZkt4GfGMBDpZnyaX3yXqG2UePQJxpM

Если мы в конструкторе класса ключа не указали параметров, то каждый раз создается новый (почти наверняка пустой – без баланса) адрес. Генерация происходит локально (без обращения к онлайн сервисам) и очень быстро. Но, если приватный ключ не сохранен, то после завершения программы доступ будет утерян. Поэтому сгенерируем приватный ключ и запишем его в блокноте. Адрес получается по свойству k.address, а приватный ключ можно получить в разных форматах, самый удобный из них – WIF (Wallet Export Format) – получаем строку методом k.to_wif():

from bit import PrivateKeyTestnet as Key

k = Key()

print('Private key:', k.to_wif())
print('Public address:', k.address)

# Private key: cQqh9xFys2KJyWhHMaBwG2kFLCNBCmTgxVqnPTXK6Vng4vU6igoV
# Public address: mhnmzFN5gr6gvmEr1t8UcRh6rdTh6JxuDe

Также по приватному ключу можно получить еще SegWit адрес. Если очень кратко, то этот адрес будет работать быстрее, чем традиционный.

print(k.segwit_address)  # 2MsWNuzx8EfgEeGLesLmkMM6q3kajEjVnVh

Воспользуемся биткоин краном, чтобы получить немного тестовых монет бесплатно:

Биткоин кран

Транзакция займет некоторое время (минут 10-20). Так что наберитесь терпения!

А пока она идет, создадим класс ключа уже из сохраненной секретной строки:

from bit import PrivateKeyTestnet as Key

k = Key('cQqh9xFys2KJyWhHMaBwG2kFLCNBCmTgxVqnPTXK6Vng4vU6igoV')
print(k.address)  # mhnmzFN5gr6gvmEr1t8UcRh6rdTh6JxuDe ура тот же!

Приватный ключ может быть представлен, как число, байты, HEX-строка, в WIF, DER и PEM форматах:

from bit import PrivateKeyTestnet as Key

k = Key('cQqh9xFys2KJyWhHMaBwG2kFLCNBCmTgxVqnPTXK6Vng4vU6igoV')

print('Int:', k.to_int(), end='\n\n')
print('Hex:', k.to_hex(), end='\n\n')
print('Bytes:', k.to_bytes(), end='\n\n')
print('WIF:', k.to_wif(), end='\n\n')
print('DER:', k.to_der(), end='\n\n')
print('PEM:', k.to_pem(), end='\n\n')

Вывод:

Int: 4397583691621789343100573085...453641742227689755261559235

Hex: 6139710fb66e82b7384b868bda1ce59a0bd216e89b8808ae503c5767e4d461c3

Bytes: b'a9q\x0f\xb6n\x82\xb78K\x86\x8b\xd...d4a\xc3'

WIF: cQqh9xFys2KJyWhHMaBwG2kFLCNBCmTgxVqnPTXK6Vng4vU6igoV

DER: b'0\x81\x84\x02\...xb3b\x8e\x1ar\xc6'

PEM: b'-----BEGIN PRIVATE KEY-----\nMIGEA.....O\nrRnD/Ls2KOGnLG\n-----END PRIVATE KEY-----\n'

Также, удобно создавать класс ключа из WIF формата функцией wif_to_key, она сама определит тип сети и создаст нужный класс:

from bit import wif_to_key

k = wif_to_key('cQqh9xFys2KJyWhHMaBwG2kFLCNBCmTgxVqnPTXK6Vng4vU6igoV')
print(k)  # <PrivateKeyTestnet: mhnmzFN5gr6gvmEr1t8UcRh6rdTh6JxuDe>

Надеюсь монеты с крана вам уже дошли, и мы продолжим.

Баланс

Узнаем баланс нашего кошелька. Для этого внутри bit используются онлайн сервисы (https://insight.bitpay.com, https://blockchain.info, https://smartbit.com.au). Поэтому операция не моментальная.

from bit import PrivateKeyTestnet as Key

k = Key('cQqh9xFys2KJyWhHMaBwG2kFLCNBCmTgxVqnPTXK6Vng4vU6igoV')
print(k.get_balance())  # 1391025

Как видите, на тот момент на адресе лежало 1391025 сатоши. 1 сатоши = одна стомиллионная целого биткоина (10-8) – самая маленькая неделимая частичка. Библиотека bit удобна еще тем, что содержит встроенный конвертер валют, поэтому баланс можно получить в любой поддерживаемой валюте: хоть в милибиткоинах, хоть в долларах, хоть в рублях. Просто передайте название валюты аргументом:

print(k.get_balance('mbtc'), 'MBTC')  # 13.91025 MBTC
print(k.get_balance('usd'), 'USD')  # 129.84 USD
print(k.get_balance('rub'), 'RUB')  # 8087.35 RUB

Как послать монеты?

Очень просто: методом send. Создадим еще один ключ (dest_k) и пошлем ему часть биткоинов от source_k:

from bit import PrivateKeyTestnet as Key

source_k = Key('cQqh9xFys2KJyWhHMaBwG2kFLCNBCmTgxVqnPTXK6Vng4vU6igoV')
dest_k = Key('cP2Z27v1ZaBz3VQRRSTQRhgYt2x8BtcmAL9zi2JsKaDBHobxj5rx')

print(f'Send from {source_k.address} to {dest_k.address}')

r = source_k.send([
    (dest_k.address, 0.0042, 'btc')
])

print(r)  # ID транзакции

Как вы помните, у транзакции может быть много выходов, поэтому первый аргумент функции send – список – кому и сколько мы посылаем (кортеж: адрес, количество, валюта). В данном случае адресат у нас один ‘n2R8xiqs6BqdgtqpXRDLKrN4BLo9VD171z’, а второй неявный выход – обратно наш же адрес, чтобы получить сдачу. Вот эта транзакция выглядит так:

Через 5 минут я уже получил первое подтверждение перевода! Проверим список транзакций:

transactions = source_k.get_transactions()
print(transactions)

# ['a101ad526e9fb131b90aac220b8b6e8bf11b9b9848ab8ea6d4384dc5b4ccece0', '0770f10a7b130852e38d9af44e050c9188664c12f2d31a56a62d6648a73e1264']

# Непотраченные входы:
unspents = source_k.get_unspents()
print(unspents)

# [Unspent(amount=967861, confirmations=4, script='76a91418ee4d98c345db083114990baa17d02e988cfedb88ac', txid='a101ad526e9fb131b90aac220b8b6e8bf11b9b9848ab8ea6d4384dc5b4ccece0', txindex=1, segwit=False)]

Пример для нескольких адресатов (каждая валюта будет пересчитана по курсу в биткоин):

my_key.send([
    ('1HB5XMLmzFVj8ALj6mfBsbifRoD4miY36v', 0.0035, 'btc'),
    ('1Archive1n2C579dMsAu3iC6tWzuQJz8dN', 190, 'jpy'),
    ('129TQVAroeehD9fZpzK51NdZGQT4TqifbG', 3, 'eur'),
    ('14Tr4HaKkKuC1Lmpr2YMAuYVZRWqAdRTcr', 2.5, 'cad')
])

Если вернуть сдачу не себе, а на другой адрес – аргумент leftover:

key.send(..., leftover='адрес_для_сдачи')

Если нужно прикрепить к транзакции сообщение (до 40 байт в кодировке UTF-8) – аргумент message:

key.send(..., message='За кофе и пончик')

Функция create_transaction только создает транзакцию и подписывает ее ключом, но не посылает ее в сеть. Аргументы те же, что у send.

Еще раз хочу подчеркнуть, что это безопасно, потому что вы отправляете уже подписанную транзакцию, перехват которой не даст злоумышленникам доступ к вашим средствам или прав на смену получателя.

В худшем случает транзакция не дойдет до сети и не исполнится.

Комиссии

Если комиссия не указана явно, то она рассчитывается по средним значениям с учетом длины транзакции. Средняя комиссия берется из онлайн-сервиса. Но можно указать комиссию самостоятельно:

# комиссия за байт (будет умножена на кол-во байт)
source_k.create_transaction(..., fee=72)  

# комиссия за всю транзакцию целиком
source_k.create_transaction(..., fee=200, absolute_fee=True)  

Полезные константы комиссий:

from bit.network import fees
fees.DEFAULT_FEE_FAST   # 10 мин
fees.DEFAULT_FEE_HOUR   # 1 час

Советы

Иногда лучше пользоваться сервисом для комиссий, потому что из-за смены нагрузки на сеть комиссия для быстрого перевода может варьироваться в широком диапазоне.

Работа с внешними онлайн сервисами может тормозить. В принципе можно поменять их приоритет, но это тема отдельной статьи. Очень хорошо, если вам доступна своя нода биткоин, тогда вы можете подключиться к ней и выполнять все действия без лишних задержек и ограничений! Вот так:

from bit.network import NetworkAPI

# тестовая нода
NetworkAPI.connect_to_node(user='user', password='password', host='localhost', port='18443', use_https=False, testnet=True)

# подключение к ноде главной сети
NetworkAPI.connect_to_node(user='user', password='password', host='domain', port='8332', use_https=True, testnet=False)

# на выбор или вместе

Храните надежно ваши приватные ключи!

Если вы брали тестовые монеты с крана, пожалуйте, верните обратно их на адрес, который вам предложат. Они могут быть полезны другим разработчикам!

Библиотека bit умеет еще работать с мульти-адресами, которые требует 2 и более подписей для выполнения транзакции.

Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈 

Библиотека schedule – CRON на Python

Вам приходилось работать с CRON? Это такой сервис в nix-системах, который позволяет регулярно в определенные моменты времени запускать скрипты или программы. Штука с долгой историей, в наследство которой достался странный синтаксис для описания правил:

0 * * * * my_script

Что если бы мы хотели иметь свой CRON внутри программы Python, чтобы в нужные моменты времени вызывать функции? Да еще, чтобы у него был человеческий синтаксис? Такая библиотека есть и называется schedule.

pip install schedule

Рассмотрим пример:

import schedule
import time

def job():
    print("Работаю")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().minute.at(":17").do(job)

# нужно иметь свой цикл для запуска планировщика с периодом в 1 секунду:
while True:
    schedule.run_pending()
    time.sleep(1)

Как видите, правила для задания временных интервалов прекрасно читаются, словно они предложения на английском языке. Перевод пары примеров:

# спланируй.каждые(10).минут.сделать(работу)
schedule.every(10).minutes.do(job)

# спланируй.каждый().день.в(10:30).сделать(работу)
schedule.every().day.at("10:30").do(job)

В задания можно передавать параметры вот так:

def greet(name):
    print('Hello', name)

schedule.every(2).seconds.do(greet, name='Alice')

Если по какой-то причине нужно отменить задание, это делается так:

def job1():
    # возвращаем такой токен, и это задание снимается с выполнения в будущем
    return schedule.CancelJob

schedule.every().day.at('22:30').do(job1)

Если нужно отменить группу заданий, то к ним добавляют тэги:

schedule.every().day.do(greet, 'Monica').tag('daily-tasks')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks')

schedule.clear('daily-tasks')  # массовая отмена по тэгу

Метод to позволяет задать случайный интервал для выполнения задания, например от 5 до 10 секунд:

schedule.every(5).to(10).seconds.do(my_job)

Библиотека сама не обрабатывает сама исключения в ваших задачах, поэтому, возможно, понадобится создать подкласс планировщика, как в этом примере. Или декоратор, который будет отменять работу, если произошло исключение. Вот так:

import functools

# декоратор для ловли исключений
def catch_exceptions(cancel_on_failure=False):
    def catch_exceptions_decorator(job_func):
        @functools.wraps(job_func)
        def wrapper(*args, **kwargs):
            try:
                return job_func(*args, **kwargs)
            except:
                import traceback
                print(traceback.format_exc())
                if cancel_on_failure:
                    return schedule.CancelJob
        return wrapper
    return catch_exceptions_decorator

@catch_exceptions(cancel_on_failure=True)
def bad_task():
    # даст исключение, но декоратор просто отменит эту задачу
    return 1 / 0  

schedule.every(5).minutes.do(bad_task)

Если задания занимают продолжительное время или должны выполняться параллельно, то вам самостоятельно придется организовать их выполнение в отдельных потоках.

import threading
import time
import schedule

# код задания
def job():
    print("Выполняюсь в отдельном потоке")


def run_threaded(job_func):
    job_thread = threading.Thread(target=job_func)
    job_thread.start()


schedule.every(10).seconds.do(run_threaded, job)
schedule.every(10).seconds.do(run_threaded, job)
schedule.every(10).seconds.do(run_threaded, job)
schedule.every(10).seconds.do(run_threaded, job)
schedule.every(10).seconds.do(run_threaded, job)


# бесконечный цикл, проверяющий каждую секунду, не пора ли запустить задание 
while 1:
    schedule.run_pending()
    time.sleep(1)

Celery

Если вы используете в проекте Celery, то, вероятно, вам не нужен schedule. В Celery и так есть отличный CRON:

from celery import Celery
from celery.schedules import crontab

app = Celery()

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # эта функция выполнится при запуске - настроим вызовы задачи test

    sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')
    sender.add_periodic_task(30.0, test.s('world'), expires=10)

    # в 7 30 по понедельникам
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test.s('Happy Mondays!'),
    )

@app.task
def test(arg):
    print(arg)

Подробнее тут.

Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈