Skip to content

Instantly share code, notes, and snippets.

View MtkN1's full-sized avatar

machuken MtkN1

View GitHub Profile
@MtkN1
MtkN1 / querying-symbol-combinations-using-sql.ipynb
Last active February 15, 2023 08:22
Querying symbol combinations using sql
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
@MtkN1
MtkN1 / python
Created January 13, 2023 07:13
Docker Run Aliases
#!/usr/bin/env bash
set -eux
docker run --rm -it \
-u vscode \
-w /home/vscode \
mcr.microsoft.com/devcontainers/python:0-${1:-3.11-bullseye} \
bash
@MtkN1
MtkN1 / minkabu_ws.py
Created December 6, 2022 02:59
みんかぶ WebSocket Sample
import socketio
sio = socketio.Client()
usdjpy = {}
@sio.event
def connect():
sio.emit("subscribe", "USDJPY")
import asyncio
from typing import List
import pybotters
import streamlit as st
async def watch_event(
store: pybotters.BybitUSDTDataStore, name: str, result: List, event: asyncio.Event
):
import asyncio
import os
import sys
import aiohttp
from loguru import logger
logger.remove()
logger.add(
sys.stdout,
@MtkN1
MtkN1 / test_oanda.py
Last active August 25, 2023 01:38
Oanda WebSocket Sample
import asyncio
import json
import aiohttp
from rich.pretty import pprint
async def main():
async with aiohttp.ClientSession() as session:
async with session.ws_connect(
@MtkN1
MtkN1 / spot.ipynb
Last active June 20, 2022 07:45
Coingecko APIでBTC/JPY市場の取引量(USD)を取得するコード
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
@MtkN1
MtkN1 / tmp.50t2wAqcq3.ipynb
Last active June 17, 2022 05:05
bitbank Socket.IO ライブラリを用いたエンコードとパース
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
@MtkN1
MtkN1 / tmp.SLXhFVB0aw.ipynb
Last active June 17, 2022 05:04
FTXの販売所レート取得
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
import asyncio
import aiohttp
import datetime
from aiohttp import web
heartbeat = 10.0
status = {}
async def ping(ws: aiohttp.ClientWebSocketResponse, specific_ping: str):
await asyncio.sleep(heartbeat)