<jupyter_start><jupyter_text>hello magnet jupyterアプリケーションを開発する上で、jupyter上で実行するためのサンプルコードを紹介します。 注意- 本jupyterファイルはサンプル紹介のため、個人の検証用ファイルは別途作成してください。- 検証のため、セキュリティ突破やpandasで利用しやすくするためのユーティリティを紹介していますが、あくまで検証用コードとして利用してください。<jupyter_code>%load_ext autoreload
%autoreload 2
import datetime
import pandas as pd
import matplotlib.pyplot as plt
from pydantic import parse_obj_as
from pprint import pprint
from magnet import get_db, Linq


def get_ohlc():
    for db in get_db():
        rep = crud.CryptoOhlcDaily(db=db)
        result = [parse_obj_as(schemas.Ohlc, x).dict() for x in rep.query()]
        # schemas.Ohlc.from_orm_query_as_dict(rep.query())
    return result
from magnet.datastore import crud, schemas

result = get_ohlc()
# print(schemas.Ohlc.__fields__["close_time"])
# print(schemas.Ohlc.from_orm(result))
for item in result:
    if item["t_cross"] == 0:
        item["t_cross_golden"] = None
        item["t_cross_dead"] = None
    elif item["t_cross"] == 1:
        item["t_cross_golden"] = item["close_price"]
        item["t_cross_dead"] = None
    elif item["t_cross"] == -1:
        item["t_cross_golden"] = None
        item["t_cross_dead"] = item["close_price"]


df = pd.DataFrame(result)[-365:]
plt.figure(figsize=(20, 10))
plt.grid()

plt.plot(df["close_time"], df["t_sma_5"], label="sma_5")
plt.plot(df["close_time"], df["t_sma_25"], label="sma_25")
plt.plot(df["close_time"], df["t_cross_golden"], marker="^", color="red", label="buy")
plt.plot(df["close_time"], df["t_cross_dead"], marker="v", color="blue", label="sell")
plt.legend(fontsize=16)
plt.show()
# matplotlib
price = [100, 250, 380, 500, 700]
number = [1, 2, 3, 4, 5]

plt.plot(price, number)
plt.title("price / number")
plt.xlabel("price")
plt.ylabel("number")

# 表示する
plt.show()
prices = [x for x in map(lambda x: x["close_price"], result)]
# print(prices)


def calc_recent_sma(arr, min=1, span1=5, span2=25, reverse=False):
    result1 = 0
    result2 = 0
    
    for index in range(span1):
        result1 += arr[index]
        
    for index in range(span2):
        result2 += arr[index]
    
    return result1 / span1, result2 / span2

def calc_target_price(arr, min=1, span1=5, span2=25, reverse=False):
    calc_arr = arr[:25]
    
    sma_5, sma_25 = calc_recent_sma(prices)

sma_5, sma_25 = calc_recent_sma(prices)
print(f"{sma_5}, {sma_25}")

# def func(**kwargs):
#     print(kwargs)

def func(from):
print(kwargs)

dic1 = {"from": "bob", "age": 20}
# dic2 = {"name": "mary"}

func(**dic1)

dic1 = {"name": "bob", "age": 20}
dic2 = {"name": "mary"}


dic1.get

def func(name, age, **kwargs):
    print(f"{name}: {age}")


func(**dic1, **dic2)
!pip install pydantic
from fastapi.encoders import jsonable_encoder
import datetime
import json
from pydantic import BaseModel, parse_obj_as, validator

class A(BaseModel):
    name: str
    dt: datetime.datetime
        
    @validator("dt")
    def valid_entry(cls, v, values, **kwargs):
        print(v)
        print(values)
        print(kwargs)
        return v

        

decoded = A(dt=datetime.datetime.now(), name="test")
decoded

print(None==0)
from typing import Union
from pydantic import BaseModel, version

print(version.version_info())

class A(BaseModel):
    name: str = "test"


obj = A()
obj = {"name": "test"}
getattr(obj, "name")
    

# a = {"name": "test", "parent": "mary"}
# # b = {"name": "bob", "age": 20}

# b = a.copy()

# b["name"] = "mary"

# print(a)
# print(b)

import inspect

a = "a"
# inspect.getmembers(a)

a += "test"

print(a)
src = {"name": "bob", "age": 20, "nest": {"name": "mary", "age": 30}}

# 方法１（シャローコピー）
# copyメソッドと同様。copyメソッドを利用しましょう。
copy1 = dict(**src)

# 方法２（シャローコピー）
# copyメソッドと同様。copyメソッドを利用しましょう。
copy1 = dict(src)

# 方法３（シャローコピー）
copy1 = src.copy()

# 方法４（ディープコピー）
import copy
copy2 = copy.deepcopy(src)

# 結果確認
print(src)
# => {"name": "bob", "age": 20, "nest": {"name": "mary", "age": 30}}

copy1["age"] += 5
copy1["nest"]["age"] += 100
print(src)
# => {"name": "bob", "age": 20, "nest": {"name": "mary", "age": 130}}

print(copy1)
# => {"name": "bob", "age": 25, "nest": {"name": "mary", "age": 130}}

copy2["nest"]["age"] -= 30
print(src)
# => {"name": "bob", "age": 20, "nest": {"name": "mary", "age": 130}}

print(copy2)
# => {"name": "bob", "age": 20, "nest": {"name": "mary", "age": 0}}
a = dict(name="test", age=20)

dict(name="val1", key2="val2", **a)
from dataclasses import dataclass
from pydantic import BaseModel

# @dataclass
class Parent(BaseModel):
    name: str
    age: int = 0

# @dataclass
class Child(BaseModel):
    name: str
    age: int = 0
    parent: User = User(name="parent", age=40)

        
a = Child(name="child", age=20)
b = Child(name="child", age=20)

print(id(a.parent) == id(b.parent))

g = {}
l = {}

result = exec("""
def detect_t_cross(topic):
    # サイン検出
    if topic["ticker"].t_cross == 0:
        return None
    elif topic["ticker"].t_cross == 1:
        return "ask"
    elif topic["ticker"].t_cross == -1:
        return "bid"
    else:
        raise Exception()
""", g, l)
print(l)
import json

s = """
{
  "obj1": {
    "name": "bob",
    "age": 20
  },
  "obj2": {
    "name": "mary",
    "age": 30
  }
}
"""

dic = json.loads(s)

# for person_id, person in dic.iteritems():
#     print(person_id, person)

    
dic = {"name": "bob"}
# dic.iteritems()
it = dic.items()
# it = it.__iter__()

# for key, value in it:
#     print(1)

# for key, value in it:
#     print(2)
print(it)
# class MyDict(dict):
#     def __setitem__(self, key, value):
#         if key in self:
#             raise KeyError(f"{key} is already exists.")
#         super().__setitem__(key, value)


# dic = MyDict({"a": 1})
# dic["a"] = 3

# print(dic)


a = {"b": 10}
a.setdefault("b", 20)

import json
import datetime
import pytz
from pytz import timezone
from pydantic import BaseModel, datetime_parse

class A(BaseModel):
    dt: datetime.datetime = None
    d: datetime.date = None
#     class Config:
#         json_encoders = {
#             datetime.datetime: lambda v: v.isoformat(),
#             datetime.date: lambda v: datetime.datetime(year=v.year, month=v.month, day=v.day).isoformat(),
#         }

# '{"dt": "2020-11-17T13:51:35.460441+00:00", "d": "2020-11-17"}'
# a = A(dt=datetime.datetime.now(tz=datetime.timezone.utc), d=datetime.datetime.now(tz=timezone("Asia/Tokyo")))
a = A(dt=datetime.datetime.now(tz=timezone("Asia/Tokyo")))
# a = A(dt=datetime.datetime.now(tz=timezone("Asia/Tokyo")), d=datetime.datetime.now(tz=timezone("Asia/Tokyo")))


str_json = a.json()
str_json
# dic = json.loads(str_json)
# dic
# obj = A(**dic)
# obj

# d = datetime.datetime.now(tz=datetime.timezone.utc)
# d.tzinfo


# d = datetime.datetime(year=2000, month=1, day=1, hour=0, tzinfo=timezone("Asia/Tokyo"))
# # d.isoformat()
# d.astimezone(datetime.timezone.utc)
# # d.date()

# d = datetime.date.today()
# d.tzinfo

# def is_aware(d):
#     if d.tzinfo is None:
#         return False
# #     if d.tzinfo.utcoffset(None) is None:
# #         return False
#     return True

# # a = datetime.datetime.now(tz=timezone("Asia/Tokyo"))
# a = datetime.datetime.now(tz=datetime.timezone.utc)

# a
# #is_aware(a)
#b = a.tzinfo.utcoffset(datetime.datetime.now())
#b = a.tzinfo.utcoffset(None)
#print(b)
# print(a.tzinfo)


# class A(BaseModel):
#     # dt: datetime.datetime
#     d: datetime.date = None

# obj = A(d=d)
# obj

# '{"dt": "2020-11-17T15:47:37.141113", "d": "2020-11-18"}'
# datetime_parse.parse_datetime("2020-11-17T15:47:37.141113")
# datetime.datetime(year=2020, month=1, day=1, tzinfo=datetime.timezone.utc)
# datetime.datetime.now(tz=datetime.timezone.utc)


tz = pytz.timezone('Asia/Tokyo')
d = datetime.datetime(year=2020, month=1, day=1, hour=9, minute=10, second=11, microsecond=123456,
                                     tzinfo=tz)
# datetime.datetime(2020, 1, 1, 9, 10, 11, 123456, tzinfo=<DstTzInfo 'Asia/Tokyo' LMT+9:19:00 STD>)
d.
tz.localize(d)
import datetime
import pytz

tz_utc = pytz.timezone("UTC")
tz_tokyo = pytz.timezone("Asia/Tokyo")

dt1 = tz_utc.localize(datetime.datetime(2010, 1, 1, tzinfo=))
dt2 = datetime.datetime(2010, 1, 1)

dt1 == dt2

import datetime
import pytz

tokyo = pytz.timezone("Asia/Tokyo")
dt = tokyo.localize(datetime.datetime(2010, 1, 1))
dt = pytz.timezone("UTC").localize(dt)
dt.date()
# datetime.datetime(2010, 1, 1, 0, 0, tzinfo=<DstTzInfo 'Asia/Tokyo' JST+9:00:00 STD>)
"test".title()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1865778200856735, 'fraction_numerical': 0.039386006663493574, 'mean_word_length': 3.961629279811098, 'pattern_counts': {'":': 54, '<': 7, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '297287', 'license': 'MIT', 'n_tokens_mistral': 3754, 'n_tokens_neox': 3486, 'n_words': 708, 'path': 'jupyter/ohlc.ipynb', 'repo_name': 'sasano8/magnet-migrade', 'size': '8404'}